collect temp files is too slow

[benbowen]

@oruebel, If I parallelize with multiprocessing this loop do you think that will cause problems? It seems to be an element of an hdf5 file. Each element is written to by an index.

[oruebel]

I don't think multiprocessing will work easily for this loop, mainly because of the call to .create_dataset(..). The read part of the loop would be fine but the data write part is problematic and I don't think you would want to load the data from all the files into memory first before you do the write.. You could probably parallelize this with MPI. In general, I think using MPI might be the better option for parallelizing things than multiprocesing.

…

On Wed, Apr 26, 2017 at 11:32 AM, Ben Bowen ***@***.***> wrote: @oruebel <https://github.com/oruebel>, If I parallelize with multiprocessing this loop <https://github.com/biorack/pactolus/blob/master/pactolus/score_frag_dag.py#L1543> do you think that will cause problems? It seems to be an element of an hdf5 file. Each element is written to by an index. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AKfYJedf53KMQQSVyVI6Yl37DznxJMKkks5rz43UgaJpZM4NJQZT> .

-- Oliver Rübel Computer Research Scientist 1 Cyclotron Road Mail Stop 50F1650 Berkeley, CA 94720 Phone: +1 510 486 4064 Email: oruebel@lbl.gov Wep: http://vis.lbl.gov/~oruebel/

[benbowen]

On one node of Edison, I can score ~10k spectra in a minute or two, but collecting the results from /tmp to $SCRATCH takes an hour or two. Scoring with serial version on 10k spectra takes an hour or two, but doesn't require collection. I'll try working with the serial version until you have time to help me with the MPI.

[oruebel]

If you don't want to do the collect as part of the parallel job, then I think you should be able to just omit the --save option so that all the scores would just end up in the temp files without copying all the data at the end to a single file. Also, if you don't need the match_matrix, then setting --match_matrix to False should help speed things up quite a bit. In general, the problem is that the code uses only serial HDF5 right now. To avoid the collect at the end in parallel, we'd need to update the code to support parallel HDF5.

…

On Wed, Apr 26, 2017 at 12:31 PM, Ben Bowen ***@***.***> wrote: On one node of Edison, I can score ~10k spectra in a minute or two, but collecting the results from /tmp to $SCRATCH takes an hour or two. Scoring with serial version on 10k spectra takes an hour or two, but doesn't require collection. I'll try working with the serial version until you have time to help me with the MPI. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AKfYJWS1D3KbJhXgoEMf1Z_-oA2Jr6ADks5rz5ukgaJpZM4NJQZT> .

-- Oliver Rübel Computer Research Scientist 1 Cyclotron Road Mail Stop 50F1650 Berkeley, CA 94720 Phone: +1 510 486 4064 Email: oruebel@lbl.gov Wep: http://vis.lbl.gov/~oruebel/

[benbowen]

I've got the serial version working. It hasn't finished searching a file yet so I don't know if the output is going to blow up.

results = sfd.score_main(use_command_line=False,
                     ms1_mass_tolerance=0.01,
                     ms2_mass_tolerance=0.01,
                     max_depth=3,
                     neutralizations=[-1.00727646677,-2.0151015067699998,0.00054857990946],
                     schedule='',
                     trees=tree_file,
                     metabolite_database=None,
                     input_filepath=container_file,
                     input_grouppath=sample_file,
                     output_filepath=out_file,
                     output_grouppath=sample_file,
                     match_matrix=False,
                     precursor_mz=-1,
                     pass_scanmeta=True,
                     pass_scans=True,
                     loglevel='ERROR', #use ERROR,INFO, or DEBUG
                     tempdir='/project/projectdirs/metatlas/projects/pactolus_tempdir',
                     clean_tempdir=True,
                     clean_output=True,
                     pass_compound_meta=False,
                     start_barrier=False)

Does wrapping the call to 'score_scan_list_against_trees' in a multiprocessing block seem valid?

[oruebel]

score_scan_list_against_trees is just a thin wrapper function that dynamically decides whether to call score_scan_list_against_trees_serial or score_scan_list_against_trees_parallel. The score_scan_list_against_trees_paralllel is effectively a wrapper for the serial function to parallelize the execution (in that case using MPI, but you could do the same with multiprocessing). The crux of it all is not to parallelize the scoring but what you need to do to collect and process all the results at the end, which is what collect_score_scan_list_results does right now. What would need to change is that score_scan_list_against_trees_paralllel should not write data to disk but do the collect through MPI data exchange. I'd need to have a look at collect_score_scan_list_results to really tell what we'd have to do to make this all work. I think this is all doable but it probably would take a couple of days of focused work to make it all work as intended. How large is the total output from the scoring right now. If I remember correctly, once you store the match matrix this becomes quite large and might not fit into memory of a single core.

…

On Wed, Apr 26, 2017 at 2:47 PM, Ben Bowen ***@***.***> wrote: I've got the serial version working. It hasn't finished searching a file yet so I don't know if the output is going to blow up. results = sfd.score_main(use_command_line=False, ms1_mass_tolerance=0.01, ms2_mass_tolerance=0.01, max_depth=3, neutralizations=[-1.00727646677,-2.0151015067699998,0.00054857990946], schedule='', trees=tree_file, metabolite_database=None, input_filepath=container_file, input_grouppath=sample_file, output_filepath=out_file, output_grouppath=sample_file, match_matrix=False, precursor_mz=-1, pass_scanmeta=True, pass_scans=True, loglevel='ERROR', #use ERROR,INFO, or DEBUG tempdir='/project/projectdirs/metatlas/projects/pactolus_tempdir', clean_tempdir=True, clean_output=True, pass_compound_meta=False, start_barrier=False) Does wrapping the call to 'score_scan_list_against_trees <https://github.com/biorack/pactolus/blob/master/pactolus/score_frag_dag.py#L1955>' in a multiprocessing block seem valid? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AKfYJZGKNH6fw24I-o5jUQHnaT8rzfciks5rz7uPgaJpZM4NJQZT> .

-- Oliver Rübel Computer Research Scientist 1 Cyclotron Road Mail Stop 50F1650 Berkeley, CA 94720 Phone: +1 510 486 4064 Email: oruebel@lbl.gov Wep: http://vis.lbl.gov/~oruebel/

[benbowen]

Thanks, becoming clear now. The temp file from a single serial process is 8.7 GB

[benbowen]

Just to close up some thoughts since it seems like this will be tabled for a while, it seems like score_scan_list_against_trees_serial is missing some data cleaning that score_scan_list_against_trees_parallel has. When we tackle this, it would be nice to have these resolved.

[oruebel]

What data cleaning do you think is missing. The output of score_scan_list_against_trees should look the same independent of whether the serial or parallel version is used. The parallel version just calls the serial version. Any further processing of results should happen in "collect_score_scan_list_results." The final output should be identical whether you run in parallel or serial, i.e., only the data that ends up in temp should be different in that the more cores you use the more files you should see in temp.

…

On Wed, Apr 26, 2017 at 4:17 PM, Ben Bowen ***@***.***> wrote: Just to close up some thoughts since it seems like this will be tabled for a while, it seems like score_scan_list_against_trees_serial is missing some data cleaning that score_scan_list_against_trees_parallel. When we tackle this, it would be nice to have these resolved. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AKfYJZb5rY1EvkKlLnH0X-UaMU7RNMR6ks5rz9CdgaJpZM4NJQZT> .

-- Oliver Rübel Computer Research Scientist 1 Cyclotron Road Mail Stop 50F1650 Berkeley, CA 94720 Phone: +1 510 486 4064 Email: oruebel@lbl.gov Wep: http://vis.lbl.gov/~oruebel/

[benbowen]

I'm not sure, but the spectra in the serial version can't be recovered from the temp file and in the parallel version they can.

I added a checker to collect_score_scan_list_results to remedy, but assumed the problem is upstream

scan_group['peak_mz'] = np.concatenate(tuple([ri[0] for ri in scan_list if len(ri)>0]), axis=-1)
scan_group['peak_value'] = np.concatenate(tuple([ri[1] for ri in scan_list if len(ri)>0]), axis=-1)

I added the len(ri)>0 and will commit tomorrow once I do some more tests. >1 would work too, but I was curious that it was an empty scan or now.

[oruebel]

That's strange. What error are you getting? I only took a quick look, but from what I can tell the list of scans is not recovered from temp files but really only passed through, i.e., the scan_list is set as input to collect_score_scan_list_results and should be identical to the scan_list that is passed into the scoring functions. That is, the behavior should be identical in serial and in parallel for this. Can you send me the script you are running and the command line output so that I can see which error you are getting and which input options you are using. Cheers, Oliver

…

On Wed, Apr 26, 2017 at 7:10 PM, Ben Bowen ***@***.***> wrote: I'm not sure, but the spectra in the serial version can't be recovered from the temp file and in the parallel version they can. I added a checker to collect_score_scan_list_results to remedy, but assumed the problem is upstream scan_group['peak_mz'] = np.concatenate(tuple([ri[0] for ri in scan_list if len(ri)>0]), axis=-1) scan_group['peak_value'] = np.concatenate(tuple([ri[1] for ri in scan_list if len(ri)>0]), axis=-1) I added the len(ri)>0 and will commit tomorrow once I do some more tests. >1 would work too, but I was curious that it was an empty scan or now. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AKfYJZeDcS9najnXtHoiQ49gG1BGtA8Sks5rz_kmgaJpZM4NJQZT> .

-- Oliver Rübel Computer Research Scientist 1 Cyclotron Road Mail Stop 50F1650 Berkeley, CA 94720 Phone: +1 510 486 4064 Email: oruebel@lbl.gov Wep: http://vis.lbl.gov/~oruebel/

[benbowen]

Its the score_main above. I'll share with and send you a link to the input files.