Implementing Parallelization with OpenDSSDirect same as OpenDSS Version 8

[marifinarif87]

Hi all,

I am using the OpenDSSdirect.py package under the Linux version. Currently, I need to use parallelization which is implemented in OpenDSS Version 8. But, I can not do that using OpenDSSdirect.py. Here is an example function which is called to iteratively solve three different distribution system in parallel using the OpenDSS version 8.

It will be great if this functionality available in OpenDSSDirect.py

function [Loads_P_Dist, Loads_Q_Dist, DistCont] = DistribSystSimul(bus_voltages_pu, num_Bus, V_Mags, iter)

global  DSSCircuit DSSText DSSSolution  

disp(V_Mags)
%% run load flow in snap shot mode
if iter ==1,
    DSSText.Command = 'set ActiveActor=*'; % message to all the actors
    DSSText.Command = 'set mode = snap maxiterations=2'; 
    
end;
for i=1:num_Bus
    DSSText.Command='Set Parallel=Yes';
    DSSText.Command=['set ActiveActor=',num2str(i)];
    DSSCircuit.Vsources.First;      % select the first VSource (normally the substation)
    DSSText.Command='VSource.source.pu='+ string(V_Mags(i));
end

DSSText.Command='Set Parallel=Yes';
DSSSolution.SolveAll
DSSCircuit.Parallel.Wait

for i=1:num_Bus
     
    DSSText.Command=['set ActiveActor=',num2str(i)];
    Load_Dist = DSSCircuit.TotalPower;
    Loads_P_Dist(i) = -Load_Dist(1,1)/1000;
    Loads_Q_Dist(i) = -Load_Dist(1,2)/1000;
    
    if DSSSolution.Converged 

        DistCont(i) = 1;
        disp('Converged')
    else
        DistCont(i) = 0;
    end
    
end
end

[PMeira]

We exchanged some emails but I thought it was a good idea to post some extended comments here.

First thing to note is that currently we don't provide a full PM implementation. This is planned for the coming months, but note that I still do not recommend blindly using the built-in PM functions -- if your simulations take a lot of time, be sure to test some alternatives to decide what's best for your use-case. Python and other languages have multiple tools to address multi-processing. The main reason the PM functions will be (re)implemented in the DSS Extensions is compatibility with the official OpenDSS, we don't want to diverge much from it. Of course there are many scenarios that justify using them -- using the diakoptics functions is a reasonable argument to use the OpenDSS parallel functions, for example.

The following examples assumes the idea is to run some scenarios, gather results, and potentially iterate with some external algorithm based on the whole set of results, while maintaining the initial circuits (without reloading everything). If someone else finds this but only need to run multiple scenarios in parallel, with long and independent runs, a simple multiprocessing.Pool is enough. It's lightly based on the sample MATLAB code from @marifinarif87 but I took some liberties when making it a full running version.

In the following pieces of code, you can change NUM_SOLVES to a large number (e.g. 200) to simulate a longer run. In that scenario, the overhead of most solutions would be irrelevant.

There might be some bug(s) here, so be sure to test it before using it for anything serious. I may revisit this sometime in the future.

Using multiprocessing and queues

Docs: https://docs.python.org/3.8/library/multiprocessing.html

Since multiprocessing is present in every Python installation, it's a natural first choice. For short tasks, I noticed a bit more overhead than Dask.distributed, for example. I think the easiest way to keep the state and change parameters is to use queues to communicate.

The following example can be applied to OpenDSSDirect.py, DSS_Python, or the official COM implementation.

import time
import multiprocessing
import numpy as np

INIT_CMD = 'compile C:/opendss/IEEETestCases/8500-Node/Master.dss'
NUM_SOLVES = 1

def run_distrib_sim_oddpy(queue_in, queue_out):
    # For OpenDSSDirect.py
    import opendssdirect as odd
    odd.Text.Command(INIT_CMD)
    odd.Text.Command('set mode=snap')
    while True:
        num, bus_voltage_pu = queue_in.get()
        odd.Text.Command(f'VSource.source.pu={bus_voltage_pu}')
        for _ in range(NUM_SOLVES): # Solve multiple times to create an artificial delay
            odd.Solution.Solve()
        converged = odd.Solution.Converged()
        queue_out.put((
            num, 
            -complex(*odd.Circuit.TotalPower()) / 1000, 
            converged
        ))


def run_distrib_sim_dsspy(queue_in, queue_out):
    # For DSS Python (tries to be drop-in replacement to COM)
    from dss import DSS
    DSS.Text.Command = INIT_CMD
    DSS.Text.Command = 'set mode=snap'
    
    while True:
        num, bus_voltage_pu = queue_in.get()
        DSS.Text.Command = f'VSource.source.pu={bus_voltage_pu}'
        for _ in range(NUM_SOLVES): # Solve multiple times to create an artificial delay
            DSS.ActiveCircuit.Solution.Solve()
        converged = DSS.ActiveCircuit.Solution.Converged
        queue_out.put((num, -complex(*DSS.ActiveCircuit.TotalPower) / 1000, converged))


def run_distrib_sim_com(cpu, queue_in, queue_out):
    # For COM 
    import win32com.client
    DSS = win32com.client.gencache.EnsureDispatch('OpenDSSengine.DSS')

    DSS.Text.Command = INIT_CMD
    DSS.Text.Command = 'set mode=snap'
    
    # This is required with OpenDSS v8+, otherwise
    # everything will run in the first CPU core
    # (for OpenDSS v7, comment this line)
    DSS.Text.Command = f'set cpu={cpu}'
    
    # The rest is the same as in run_distrib_sim_dsspy
    
    while True:
        num, bus_voltage_pu = queue_in.get()
        DSS.Text.Command = f'VSource.source.pu={bus_voltage_pu}'
        for _ in range(NUM_SOLVES): # Solve multiple times to create an artificial delay
            DSS.ActiveCircuit.Solution.Solve()
        converged = DSS.ActiveCircuit.Solution.Converged
        queue_out.put((num, -complex(*DSS.ActiveCircuit.TotalPower) / 1000, converged))


run_distrib_sim = run_distrib_sim_oddpy

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    queue_in = manager.Queue()
    queue_out = manager.Queue()
    pool = multiprocessing.Pool()
    pool_size = len(pool._pool)

    
    # Fill the input queue with some values
    bus_voltages_pu = np.linspace(0.9, 1.1, num=128)
    for num, v in enumerate(bus_voltages_pu):
        queue_in.put((num, v))


    t0 = time.perf_counter()
    # This will start this processes
    # Each will load the circuit and then wait for new items to run
    
    # OpenDSSDirect.py or DSS_Python versions
    pool.starmap_async(
        run_distrib_sim, 
        [(queue_in, queue_out) for _ in range(pool_size)]
    )

    # COM version
    # pool.starmap_async(
        # run_distrib_sim_com, 
        # [(cpu, queue_in, queue_out) for cpu in range(pool_size)]
    # )
    
    print('Waiting...')
    while queue_out.qsize() < len(bus_voltages_pu):
        time.sleep(0.1)
        
    print(f'Total time: {time.perf_counter() - t0} s')

    results = []
    while not queue_out.empty():
        results.append(queue_out.get())
        
    print(f'Total results: {len(results)}')
    # for num, total_power, converged in results:
        # print(num, total_power, converged)

    print('Done!')
    pool.terminate()
    pool.join()

Using Dask.distributed

Docs: https://distributed.dask.org/en/latest/actors.html

As of Nov/2020, actors are currently marked as experimental in Dask.distributed but they might be useful. Even with the official COM implementation on Windows, I noticed it runs faster (10-20%) than the official PM functions from OpenDSS v8/v9, even though it uses processes instead of threads. The local cluster initialization takes some time but that shouldn't matter for long simulations.

Note that this could be extended to multiple machines by using a dedicated scheduler and set of workers. Check the Dask.distributed docs for more.

The main things to note here is that we need to use 1-thread per worker and also assign a single actor per worker.

This example uses ODD.py, but it's trivial to adapt to DSS_Python. For the official OpenDSS COM, you'd need to initialize the COM object in the actor's __init__ as well as set the CPU of the default actor on OpenDSS level (even though we're not using the PM functions, the default is to use the first core only).

import numpy as np
import opendssdirect as odd

INIT_CMD = 'compile /opendss/IEEETestCases/8500-Node/Master.dss'
NUM_SOLVES = 1

class DSSActor:
    def __init__(self):
        odd.Text.Command(INIT_CMD)
        odd.Text.Command('set mode=snap')

    def Command(self, cmd):
        odd.Text.Command(cmd)

    def Solve(self):
        for _ in range(NUM_SOLVES): # Solve multiple times to create an artificial delay
            odd.Solution.Solve()
            
        return (
            -complex(*odd.Circuit.TotalPower()) / 1000, 
            odd.Solution.Converged()
        )


if __name__ == '__main__':
    from distributed import Client
    from time import perf_counter
    
    # Create a local Dask.distributed cluster
    client = Client(threads_per_worker=1)
    print(client.cluster)

    # Create an actor per worker process
    num_actors = len(client.cluster.workers)
    actors = [
        client.submit(DSSActor, actor=True, workers=[w]).result()
        for w in client.cluster.workers
    ]

    # Run a simulation loop
    t0 = perf_counter()
    all_results = []
    bus_voltages_pu = np.linspace(0.9, 1.1, num=128)
    
    offset = 0
    while offset < 128:
        #TODO: change something for each cycle
        futures = []
        for num, v in enumerate(bus_voltages_pu[offset:(offset + num_actors)]):
            actors[num].Command(f'VSource.source.pu={v}')
            futures.append(actors[num].Solve())
            
        offset += num_actors
        all_results.extend([f.result() for f in client.gather(futures)])
    
    print(f'Total time: {perf_counter() - t0} s')
    print(f'Total results: {len(all_results)}')
    
    # print('Results:')
    # for num, (total_power, converged) in enumerate(all_results):
    #     print(num, total_power, converged)

    client.close()

Other options?

I imagine there are many other options that could handle this simulation scenario. Since I don't tend to use stateful actors/workers, I wouldn't be able to recommend many. I've seen Ray a bit but haven't tried it yet (it supports actors).

And as a reminder, when each scenario is an independent simulation, there are even more options. multiprocessing, Dask, Ray, Celery, Dramatiq, ... could all be used for local and distributed workloads.

[PMeira]

For future reference: we now have the "parallel-machine" support form OpenDSS v8+ again, with a more robust implementation than before. I still suggest users to run a few tests with multi-processing and other alternatives to check what's best for their use-cases (soon we'll multiple DSS instances in OpenDSSDirect.py, already implemented in DSS Python) .