Freeing memory in python parallel process loops
<p>I'm using a master-slaves structure to implement a parallel computation.  A single master process (<code>0</code>) loads data, and distributes relevant chunks and instructions to slave processes (<code>1</code>-<code>N</code>) which do the heavy lifting, using large objects... blah blah blah.  The issue is memory usage, which I'm monitoring using <code>resource.getrusage(resource.RUSAGE_SELF).ru_maxrss</code> on each slave process.</p>

<p>The first task uses about 6GB of memory, as expected, but when the slave receives the second task, it balloons up to just over 10GB --- as if the previous memory wasn't being collected.  My understanding was that as soon as a variable looses its references (in the below code, when the <code>_gwb</code> variable is reset) garbage collection should clean house.  Why isn't this happening?</p>

<p>Would throwing in a <code>del _gwb</code> at the end of each loop help?<br>
What about a manual call to <code>gc.collect()</code>?<br>
Or do I need to spawn <code>subprocess</code>es as <a href="http://stackoverflow.com/a/24126616/230468">described in this answer</a>?</p>

<p>I'm using <code>mpi4py</code> on a SLURM managed cluster.</p>

<p>The <strong>master</strong> process looks something like:</p>

<pre><code>for jj, tt in enumerate(times):

    for ii, sim in enumerate(sims):

        search = True
        # Find a slave to give this task to
        while search:
            # Repackage HDF5 data into dictionary to work with MPI
            sim_dat = ...  # load some data

            # Look for available slave process
            data = comm.recv(source=MPI.ANY_SOURCE, tag=MPI.ANY_TAG)
            src = stat.Get_source()

            # Store Results
            if tag == TAGS.DONE:
                _store_slave_results(data, ...)
                num_done += 1
            elif tag == TAGS.READY:
                # Distribute tasks
                comm.send(sim_data, dest=src, tag=TAGS.START)
                # Stop searching, move to next task
                search = False

            cycles += 1
</code></pre>

<p>And the <strong>slaves</strong>:</p>

<pre><code>while True:
    # Tell Master this process is ready
    comm.send(None, dest=0, tag=TAGS.READY)
    # Receive ``task`` ([number, gravPot, ndensStars])
    task = comm.recv(source=0, tag=MPI.ANY_TAG, status=stat)
    tag = stat.Get_tag()

    if tag == TAGS.START:
        _gwb = Large_Data_Structure(task)
        data = _gwb.do_heavy_lifting(task)
        comm.send(data, dest=0, tag=TAGS.DONE)
    elif tag == TAGS.EXIT:
        break

    cycles += 1
</code></pre>

<hr>

<p>Edit:  Some other strange subtleties (in case they might be relevant):<br>
1) only some processes show the memory growing, other stay roughly the same;<br>
2) The specific amount of memory active is <em>different</em> on the different slave processes (differing by <code>100s of MB</code> ... even though they should necessarily be running the same code!</p>

<p><code>del _gwb</code> should make a big difference. With <code>_gwb = Large_Data_Structure(task)</code> the new data is generated and then assigned to _gwd. Only then is the old data released. A specific <code>del</code> will get rid of the object early. You may still see a memory increase for the second loop - python releases the object into its heap but there's nothing to say that the next allocation will get exactly the same bunch of memory.</p>

<p>The garbage collector only comes into play in cases where regular reference counting isn't sufficient to trigger freeing of the memory. Assuming <code>do_heavy_lifting</code> isn't doing anything funky, it won't make a difference.</p>

<p>You mention <code>subprocess</code>... another option on linux systems is <code>os.fork</code>. The child process gets a copy-on-write view of the parent address space. The big object is generated in the child memory and that goes away on exit. I can't guarantee this will work but would be an interesting experiment.</p>

<pre><code>while True:
    # Tell Master this process is ready
    comm.send(None, dest=0, tag=TAGS.READY)
    # Receive ``task`` ([number, gravPot, ndensStars])
    task = comm.recv(source=0, tag=MPI.ANY_TAG, status=stat)
    tag = stat.Get_tag()

    if tag == TAGS.START:
        pid = os.fork()
        if pid:
            # parent waits for child
            os.waitpid(pid)
        else:
            # child does work, sends results and exits
            _gwb = Large_Data_Structure(task)
            data = _gwb.do_heavy_lifting(task)
            comm.send(data, dest=0, tag=TAGS.DONE)
            os._exit()
    elif tag == TAGS.EXIT:
        break

    cycles += 1
</code></pre>

python performance memory parallel-processing numerical-methods