I created a binary file format for storing point clouds of around 10mil points. I also store a kdtree, so I don't have to read the entire file, as I will be loading circular crops from the file. Calling this function 1 thousand times with a circle spanning around 50% of the points takes around 20s. When I use a line profiler, I see about 30% of this time is spend reading, so with multi-processing, I expect about 3 times speed-up if I use more processors. Yet adding multi-processing does not seem to help significantly (maybe 10% speed up). how come?
#🔒 Speeding up my file reading
loud stone
mellow bladeBOT
I created a binary file format for storing point clouds of around 10mil points. ...
@loud stone
Python help channel opened
loud stone
Here is the line profiler of the main read function:
mellow bladeBOT
Here is the line profiler of the main read function:
Please react with ✅ to upload your file(s) to our paste bin, which is more accessible for some users.
loud stone
And here is my apply_offset_and_scaling function, which takes up around 30% of the time:
def apply_offset_and_scaling(
raw_values: NDArray[np.integer | np.floating],
offset: float | None,
scaling: float | None,
return_dtype: np.dtype,
) -> NDArray[np.integer | np.floating]:
"""
Converts raw values to true values by applying offset and scaling.
Args:
values (NDArray):
Raw values to be converted.
offset (float):
Offset for converting raw to true values.
scaling (float):
Scaling for converting raw to true values.
return_dtype (FieldDtype, optional):
Dtype of the values.
Returns:
NDArray:
True values.
"""
values = raw_values.astype(return_dtype, copy=True)
if scaling is not None and scaling != 1:
values *= scaling
if offset is not None and offset != 0:
values += offset
return values
I created 1000 separate files. Sequentially I call it with:
for i in range(NR_REPEATS):
tmp_file = str(tmp_dir / f"test_{0}.bin")
read_func(tmp_file)
and parallel:
inputs = [str(tmp_dir / f"test_{i}.bin") for i in range(NR_REPEATS)]
with mp.Pool(processes=5) as pool:
pool.map(read_func, inputs)
Any suggestions appreciated
quick fossil
how much data is it?
you normally dont see much speed improvments if you keep things as strings
at least use bytes
loud stone
10 mil points, saved as int16, x, y and z
quick fossil
oh so they are already int16?
loud stone
They are saved in the file as int16 yes, so if you see in the line-profiler, line:
xs_inside = read_field_indices_raw(f, header_read, "x", inside)
xs_inside is a numpy array of int16
loud stone
Literally just import:
import multiprocessing as mp and then the code shown above:
inputs = [str(tmp_dir / f"test_{i}.bin") for i in range(NR_REPEATS)]
with mp.Pool(processes=5) as pool:
pool.map(read_func, inputs)
i would not use mp, i would use a more modern api, like concurrent futures
and i would not default to testing out only processpoolexecutor (the paralell feature of mp.pool)
i would also test threadpoolexecutor as well
loud stone
Hmm Alright. but still, I don't think it's the overhead of spawning new workers, I also tried adding a loop in the function itself to process 100 files, and then over 10 processes f.e.
quick fossil
doing what i talk about here is how i solved 1brc in under five seconds
but maybe you could also test out polars, see if that alone gives you what you need
loud stone
Could it just be memory bandwith?
That creating/deleting the arrays is the bottleneck
concurrent did actually speed it up by a bit more (~80%)
hard dagger
80% is a lot process pool or thread pool
glass linden
as @quick fossil asked before, how much data is it?
on disk, per file on average and in total?
hard dagger
Could it just be memory bandwith?
might wanna check ur read/write speed and match that to your drive specs if u suspect a bottleneck
loud stone
10 mil points, 50% being read, 3 columns, each 16 bit int
glass linden
yeah, but in bytes on disk
hard dagger
int 16 so 16bits * 10mil roughly ig
glass linden
oh, not a lot, that should be cached in memory by the OS without a problem after the first read if the files doesn't change
loud stone
I read separate files
hard dagger
int 16 so 16bits * 10mil roughly ig
this is about 20mb if u add a 50% overhead then maybe 30mb
loud stone
I created 1000 separate files. Sequentially I call it with:
```py
for i in r...
1000 separate files is the test^
loud stone
this is about 20mb if u add a 50% overhead then maybe 30mb
*3 (x, y, z)
hard dagger
oh right point cloud
glass linden
1000 separate files is the test^
60 MB per file or in total?
loud stone
Per file
hard dagger
fair 60-70mb makes sense per file
loud stone
Each file has those specs
hard dagger
how many such files
loud stone
I just made a copy for each file for timing my function
hard dagger
1000 i see
glass linden
oh, than that is probably too much to cache in RAM unless you have a lot of RAM
hard dagger
thats plenty
loud stone
As I am training a model on hundreds of GB of data
hard dagger
hmm i see rn 1000 files takes how long to load with concurrent
glass linden
otherwise memory mapped (mmap) files can sometimes help
loud stone
And even then, loading the data from SSD to ram only is 30% of the time spent in the function
hard dagger
i am assuming u have x workers and load x files at a time?
loud stone
Every worker loads a circle from a separate file
hard dagger
import numpy as np
from numpy.typing import NDArray
def apply_offset_and_scaling(
raw_values: NDArray[np.integer | np.floating],
offset: float | None,
scaling: float | None,
return_dtype: np.dtype,
) -> NDArray[np.integer | np.floating]:
has_scaling = scaling is not None and scaling != 1
has_offset = offset is not None and offset != 0
if not has_scaling and not has_offset:
return raw_values.astype(return_dtype, copy=False)
if has_scaling and has_offset:
return (raw_values * scaling + offset).astype(return_dtype, copy=False)
elif has_scaling:
return (raw_values * scaling).astype(return_dtype, copy=False)
else:
return (raw_values + offset).astype(return_dtype, copy=False)
what if u do something like this i think the memory allocation might be slowing things down for u?
u combine both ops to prevent intermediate allocs
loud stone
Line # Hits Time Per Hit % Time Line Contents
==============================================================
8 @profile
9 def apply_offset_and_scaling(
10 raw_values: NDArray[np.integer | np.floating],
11 offset: float | None,
12 scaling: float | None,
13 return_dtype: np.dtype,
14 ) -> NDArray[np.integer | np.floating]:
...
32
33 8000 14364495.1 1795.6 77.9 values = raw_values.astype(return_dtype, copy=True)
34 8000 9786.2 1.2 0.1 if scaling is not None and scaling != 1:
35 6000 4056231.4 676.0 22.0 values *= scaling
36
37 8000 5733.4 0.7 0.0 if offset is not None and offset != 0:
38 values += offset
39
40 8000 2625.8 0.3 0.0 return values
I create the array, there's no getting around that as I need to allocate memory for the resulting data. And then scaling takes about 20%
What you have written would actually create more copies in some cases, as every operation like + and * would need to allocate a new array for the results.
Which is why I do *= and += rn
hard dagger
but if both scaling and offset are present ur going over the array twice
loud stone
So are you 😛
Python will apply the operations sequentially
arr * 5 + 3 will go over the array twice
hard dagger
hmm can u profile mine once if possible
loud stone
Sure, 1 sec
hard dagger
thx
loud stone
Line # Hits Time Per Hit % Time Line Contents
==============================================================
43 @profile
44 def apply_offset_and_scaling(
45 raw_values: NDArray[np.integer | np.floating],
46 offset: float | None,
47 scaling: float | None,
48 return_dtype: np.dtype,
49 ) -> NDArray[np.integer | np.floating]:
50 8000 7840.9 1.0 0.0 has_scaling = scaling is not None and scaling != 1
51 8000 2888.6 0.4 0.0 has_offset = offset is not None and offset != 0
52
53 8000 3522.5 0.4 0.0 if not has_scaling and not has_offset:
54 2000 1273111.2 636.6 7.2 return raw_values.astype(return_dtype, copy=False)
55
56 6000 1324.2 0.2 0.0 if has_scaling and has_offset:
57 return (raw_values * scaling + offset).astype(return_dtype, copy=False)
58 6000 1198.2 0.2 0.0 elif has_scaling:
59 6000 16285876.9 2714.3 92.7 return (raw_values * scaling).astype(return_dtype, copy=False)
60 else:
61 return (raw_values + offset).astype(return_dtype, copy=False)
Total time: 17.5758 s
My total time was 6.55603 s for
hard dagger
hmm i see damn thats 3 times
loud stone
This is likely because your function results in float64, and then converts to float32
You can do copy=false, but it will need a copy if it's not the correct dtype (or nr bytes)
Anyways, I got about 80%, I think i'll just roll with it for now, but thx for any inputs!
hard dagger
yeah scaling is a float so entire temp arr prolly becomes float64
what if we use out param and write to a pre defined buffer https://stackoverflow.com/questions/27293830/utility-of-parameter-out-in-numpy-functions kinda very similar to what ur doing but with more explicit conditions and one main alloc
def apply_offset_and_scaling(
raw_values: NDArray,
offset: float | None,
scaling: float | None,
return_dtype: np.dtype,
) -> NDArray:
has_scaling = scaling is not None and scaling != 1
has_offset = offset is not None and offset != 0
if not has_scaling and not has_offset:
return raw_values.astype(return_dtype, copy=False)
out = np.empty_like(raw_values, dtype=return_dtype)
if has_scaling and has_offset:
np.multiply(raw_values, scaling, out=out)
np.add(out, offset, out=out)
elif has_scaling:
np.multiply(raw_values, scaling, out=out)
else:
np.add(raw_values, offset, out=out)
return out
That's a good idea, but even then, I need to allocate memory, but then it will be outside the function.
It would help me if I need to copy it to a larger array, which I might want later on
hard dagger
what if we use `out` param and write to a pre defined buffer https://stackoverfl...
thats fair, can u just profile this once I think this should do 1 alloc + 1 full read and write, compared to the older alloc + copy + r/w (avoids the astype call)
also because I am curious 
loud stone
Line # Hits Time Per Hit % Time Line Contents
==============================================================
42 @profile
43 def apply_offset_and_scaling(
44 raw_values: NDArray,
45 offset: float | None,
46 scaling: float | None,
47 return_dtype: np.dtype,
48 ) -> NDArray:
49 8000 3989.3 0.5 0.0 has_scaling = scaling is not None and scaling != 1
50 8000 1709.2 0.2 0.0 has_offset = offset is not None and offset != 0
52 8000 1748.4 0.2 0.0 if not has_scaling and not has_offset:
53 2000 1262130.7 631.1 15.5 return raw_values.astype(return_dtype, copy=False)
55 6000 38296.1 6.4 0.5 out = np.empty_like(raw_values, dtype=return_dtype)
57 6000 1878.3 0.3 0.0 if has_scaling and has_offset:
58 np.multiply(raw_values, scaling, out=out)
59 np.add(out, offset, out=out)
60 6000 1806.6 0.3 0.0 elif has_scaling:
61 6000 6807299.8 1134.5 83.8 np.multiply(raw_values, scaling, out=out)
62 else:
63 np.add(raw_values, offset, out=out)
65 6000 1319.3 0.2 0.0 return out
8.1s
hard dagger
well rip I can't think of anything else
glass linden
i haven’t used numba for years now, but maybe it might be applicable in this instance?
loud stone
Yeah, I don't expect to speed up this functioln. Just that it would be easily parallelized if called in parallel
hard dagger
!pypi numexpr might help, maybe numba jit stuff too
mellow bladeBOT
loud stone
I tried numba, it didn't made much of a difference
numexpr too, but I'm only doing scaling rn, as offset is set to 0, so this would not help for my current example
I've also rewritten parts of my other code in Cython, so i'm trying everything 😛
glass linden
but you have gotten it down to about 6.55s, which isn't that bad, is it?
loud stone
I don't care about the offset_and_scaling function being slow or not, I care abut the fuct that calling my read function in parallel does not seem to speed it up
And apply offset and scaling is a large portion of it, and seems to not be much quicker when multiple workers are calling it in parallel
glass linden
are you processing one file from start to finish per process?
hard dagger
are u using process pool executor from concurrent.futures
loud stone
Yes, that suggestion by eivl sped it up by 80% instead of 10% with multiprocessing library
But I expected a speed-up of 3 being possible, as IO is only 30% (file reading IO I mean)
glass linden
but you really should be able to use all your cores if you are processing the data from each file independently
loud stone
Yeah, I know 😛
But it's not
So my current thought it that my bottleneck is allocating memory for arrays
Since that can't be sped up with multiple workers
glass linden
but it should probably tax more than one core anyways even if it's the memory allocation that is the bottleneck
loud stone
I meant 100% 1 core when run sequentially, I thought you tried to figure out if numpy was using multiple cores
In parallel it will use 100% of a core per worker
I checked with htop
hard dagger
can u show how ur calling the process pool part
loud stone
It's in my initial message
mellow bladeBOT
Python help channel closed for inactivity