Retrieve feedback indicators from grid indexing

[loichuder]

When running grid indexing in Ewoks, we would like to provide feedback to the user on the process advancement.

The grid indexing prints some useful information with print or through sys.stderr:

ImageD11/ImageD11/grid_index_parallel.py

Line 177 in 41cd63a

sys.stderr.write(" % 6.2f%% Position %d %d %d"%(i/ni,t_x, t_y, t_z)+

One of the strategy we considered was to somehow capture this text output to display it in a text box to the user.

Unfortunately, stderr is difficult to capture so we could imagine replacing those by logging calls instead ?

Some recipes already exist for logging in a multiprocessing context

Of course, we would be happy to know if you have another idea on how to provide feedback to the user.

Thanks!

[loichuder]

@abmajith

[jonwright]

The python logging module did not mix well with multiprocessing, this is why ImageD11.indexing has a workaround. Maybe this is the origin of your problem? A huge amount of debug output from indexing needs to be suppressed when you run grid index.

I assume you plan to run the gui on a single core in a front end node and then run all the processing jobs as slurm or ewoks jobs. Then you can do tail -f to follow the stdout and stderr via GPFS. On a local machine I guess you use a python subprocess and grab the stdout and stderr from that process?

Computationally demanding functions in ImageD11 are not intended to run in the GUI process address space. Like that you have no easy way to kill them.

[loichuder]

The python logging module did not mix well with multiprocessing, this is why ImageD11.indexing has a workaround.

If I understand correctly, your technique would be to run the grid indexing in a separate program and redirect the outputs to a file ?

Something like

python grid_indexing.py > index.log

I assume you plan to run the gui on a single core in a front end node and then run all the processing jobs as slurm or ewoks jobs.

Maybe in the future but so far, the processing is running on the same machine as the GUI. So the idea would be to request an interactive job on SLURM (with X11 forwarding) and run ewoks3xrd there

[abmajith]

Hi @jonwright et @loichuder

We tried to kill on the grid index process and its subprocess, we think it will work like this,

def grid_index_worker(
    gridpars, q, async_result
):
    """
    Input Args:
      - `gridpars` : grid indexing parameters
      - `q` : Multiprocessing Queue
      - `async_result`: Multiprocessing 'MapResult' Class instance
    """

    ul = uniq_grain_list(gridpars['SYMMETRY'], gridpars['toldist'], gridpars['tolangle'])
    lastsave = 0

    while True:
        try:
            grs = q.get(True, 10)
            gb4 = len(ul.uniqgrains)
            ul.add(grs)
            gnow = len(ul.uniqgrains)
            logger.info( "Got % 5d new %d from %d", gnow, gnow-gb4, len(grs))
            if len(ul.uniqgrains) > lastsave:
                lastsave = len( ul.uniqgrains )
                grain.write_grain_file( gridpars['output_filename'], ul.uniqgrains )
            if async_result._number_left == 0:
                break
        except Queue.Empty:
            logger.info(" Caught queue empty exception\n")
            if async_result._number_left == 0:
                break
        except KeyboardInterrupt:
            break
    
    grain.write_grain_file( gridpars['output_filename'], ul.uniqgrains )

    
def grid_index_parallel_with_pool(fltfile, parfile, tmp, gridpars, translations, log_queue=None):
    """
    Input Args:
      - `fltfile`  : filtered peaks file,
      - `parfile`  : contains geometry and lattice parameters
      - `tmp`      : base folder (directory) to save intermediate and final results.
      - `gridpars` : grid indexing parameters:
        dict of following parameters:
        - `NPKS`       (int)           : number of peaks
        - `DSTOL`      (float)         : reciprocal distance tolerance
        - `RING1`      (Sequence[int]) : default (1, 0)
        - `RING2`      (Sequence[int]) : default (0,)
        - `NUL`        (bool)          : default True
        - `FITPOS`     (bool)          : default True
        - `tolangle`   (float)         : default 0.50
        - `toldist`    (float)         : default 100.0
        - `NPROC`      (int)           : default None, but ImageD11 will set it to maximum available cores on the server
        - `NTHREAD`    (int)           : default 1
        - `COSTOL`     (float)         : default None if default provided then here will set it to np.cos(np.radians(90 - 2 * omega_slop))
        - `OMEGAFLOAT` (float)         : default None if default provided then here will set it to omega_slop
            Note: omega_slop is related to wobble of the instrument,
        - `TOLSEQ` (Tuple[float, ...]) : default (0.02, 0.015, 0.01)
        - `SYMMETRY` (str)             : default cubic

    Output Args:
      - `p`           : Multiprocessing Pool Object
      - `q`           : Multiprocessing Queue Class Object instance
      - `async_result`: Multiprocessing 'MapResult' Class instance
      - `gridpars`    : initialized grid index parameter (dict of parameters)
    """

    gridpars = initgrid(fltfile, parfile, tmp, gridpars)
    logger.info( "Done init" )
    if 'NPROC' not in gridpars or gridpars['NPROC'] is None:
        NPR = max( cImageD11.cores_available() - 1, 1 )
        cImageD11.cimaged11_omp_set_num_threads(1)
    else:
        NPR = int(gridpars['NPROC'])
    if 'NTHREAD' in gridpars:
        cImageD11.cimaged11_omp_set_num_threads(int(gridpars['NTHREAD']))
    elif NPR > 1:
        cImageD11.cimaged11_omp_set_num_threads(1)
    tsplit = [translations[i::NPR] for i in range(NPR)]
    args = [("%s.flt" % tmp, parfile, t, gridpars) for i, t in enumerate(tsplit)]

    if log_queue is None:
        q = PQueue()
    else:
        q = log_queue

        
    p = Pool(processes=NPR, initializer=wrap_test_many_points_init, initargs=[q])
    logger.info(f"Using pool of {NPR} processes")
    async_result = p.map_async(wrap_test_many_points, args)
    
    return p, async_result, q, gridpars


def grid_index_parallel( fltfile, parfile, tmp, gridpars, translations ):
    """
    fltfile containing peaks
    parfile containing instrument geometry and unit cell
    tmp - base name for scratch files and results
    gridpars : dictionary of control parameters (rings to use, etc)
    translations : list of translation positions to try

    Runs a grid index algorithm using pythons multiprocessing module
    splits workload over processes (blocks of translations to each process)
    This thread should catch results via a queue
    """
    p, async_result, q, gridpars = grid_index_parallel_with_pool(fltfile, parfile, tmp, gridpars, translations)
    grid_index_worker(gridpars, q, async_result)
    p.close()
    p.join()

We are still working on how to safely put the print, stderr redirect to terminal or logger file using multiprocessing.Queue in logging module, it showed some positive result, will test them extensively, will communicate later.

[jonwright]

If you run the grid_index script from ImageD11 then I suggest you run it via the python subprocess module (https://docs.python.org/3/library/subprocess.html)

If you go for a re-write then it makes sense to also look at using MPI (e.g. concurrent.futures style interfaces).

[loichuder]

If you go for a re-write then it makes sense to also look at using MPI (e.g. concurrent.futures style interfaces).

You mean by using mpi4py ?

@abmajith is motivated to work on this but would you be open to add mpi4py as a dependency of ImageD11 then ?

[abmajith]

If you run the grid_index script from ImageD11 then I suggest you run it via the python subprocess module (https://docs.python.org/3/library/subprocess.html)

now we are using the popen subprocess to gather information from the terminal to the user for feedback.

I also understand now the grid index parallel module, If you want to do some changes or refactor it, I am in it

[jonwright]

If you go for a re-write then it makes sense to also look at using MPI (e.g. concurrent.futures style interfaces).

You mean by using mpi4py ?

For parallel access to hdf5 files I am afraid we need mpi4py. It seems we come back to #252 every time some parallel issue comes up.

To optimise grid_index itself there are various ways to improve the algorithm (e.g. look for Friedel pairs). Otherwise it might be ported to GPU somehow.