Microsoft Robotics. Parallel data processing

One of Microsoft’s products, Microsoft Robotics, includes the Concurrent and Coordination Runtime library. The library is very necessary for robots to organize parallel computing in the processing of sound and images (and not only them). In the post I will explain how using this library to create a parallel algorithm for multiplying the matrix A (m × n elements) by the vector B (1 × n elements) using the following formula: . The algorithm for calculating the vector C (m × 1 elements) is easily parallelized, since the value of the ith element of the vector does not depend on the values ​​of its other elements. Before running the example from source, it is recommended to install Microsoft Robotics.

1. A brief introduction to Microsoft Robotics

Microsoft Robotics has several components:

• Library Concurrent and Coordination Runtime (CCR) - designed to organize data processing using parallel and asynchronously running methods. The interaction between such methods is organized on the basis of messages. Mailing is based on the use of ports;
• Decentralized Software Services (DSS) - an environment that allows you to run data processing algorithms on different computers, to organize the asynchronous interaction of the management processes of various robot subsystems;
• Visual Simulation Environment (VSE) - a visualization environment that allows you to experiment with models of robots, test robots control algorithms;
• Visual Programming Language (VPL) is a language designed for developing robot control programs. A program in such a language is represented as a sequence of blocks that perform data processing and links between them.

Microsoft Robotics Core Concepts:

• service - an application developed in Microsoft Robotics;
• message - an instance of any data type;
• port - a message queue of type FIFO (First-In-First-Out), the message remains in the port until it is retrieved from the port queue by the receiver.
  Port Definition:
  

  Port<int> p = new Port<int>(); 

  
  Sending a message to the port:
  

   p.Post(1); 

  
  
• recipient - a structure that performs message processing. This structure combines:
  
  1. one or more ports to which messages are sent;
  2. the method (or methods) used to process messages (this method is called a task);
  3. logical condition that determines the situations in which a particular recipient is activated.
  
  Recipient Description:
  

   Arbiter.Receive(<_>, <>, <_>); : Activate( Arbiter.Receive(false, intPort, delegate(int n) { Console.WriteLine(" : " + n.ToString()); } ) ); 

  
  The delegate entering the receiver will be executed when a message arrives at the intPort port.
  Message recipients are of two types: temporary and permanent (in the example, the recipient is temporary). A temporary recipient, after processing a message (or several messages), is deleted from the list of message recipients of this port.
  
• the task launch process is controlled by the dispatcher. After the activation conditions of the task are fulfilled (one of the activation conditions may be the receipt of a message by the port), the dispatcher assigns the task a thread from the thread pool in which it will run.
  Description of the dispatcher with two threads in the pool:
  

   Dispatcher d = new Dispatcher(2, "MyPool"); 

  
  Description of the dispatcher's queue, in which tasks are put for execution:
  

   DispatcherQueue dq = new DispatcherQueue("MyQueue", d); 

  
  

2. Creating a service project

A service project is created in Visual Studio on the Microsoft Robotics tab (see Figure 1).


Fig. 1. Creating a service project

After creating the project, you need to configure the service: specify the name for it, the namespace, the domain of the organization for which the service is being developed, as well as the year and month of service creation. Domain and date are used to create a unique identifier for the service (in this project, this identifier will not be needed directly).
')

Fig. 2. Service setting

The service project consists of several files:

• ParallelProcessing.cs is the core of the service;
• ParallelProcessingTypes.manifest.xml - a manifest that DSS uses to load the service;
• ParallelProcessingTypes.cs - contains a description of the types that are used by the service.

After compiling the service, a set of dll libraries and a manifest are generated. Manifest is a file in XML format containing references to the services with which the developed service interacts. The service project settings are set so that after compiling the service, the created dll libraries are copied to the \ bin directory.
The service is started using the dsshost program, which loads the manifest and all the services that are listed in it.

3. Service structure

The structure of the service code includes the Start method, which is called when the DSS node (dsshost.exe) starts the service (therefore, data initialization actions are usually placed on the Start method):

 protected override void Start() { base.Start(); //   //    } 

After compiling the project, three dll libraries are generated:

• ParallelProcessing.Y2013.M02.dll is a library of service implementation formed on the basis of the source code files included in the service project;
• ParallelProcessing.Y2013.M02.proxy.dll allows you to use your service by other services;
• ParallelProcessing.Y2013.M02.transform.dll contains a description of the correspondence between the types defined in the service implementation and the proxy implementation of the library. This library is automatically loaded using the DSS runtime.

You can execute the created service not only through F5 in Visual Studio, but also via the command line. To do this, select "DSS Command Prompt" in the Start \ MRDS menu. As a result, the command interpreter window will start, the root directory and environment variables will be set. In the window that opens, run the following command for execution:

 dsshost /p:50000 /t:50001 /m:"<    >" 

4. Estimated lead time

The execution time of the calculations will be determined using the Stopwatch class:

 Stopwatch sWatch = new Stopwatch(); sWatch.Start(); < > sWatch.Stop(); Console.WriteLine(sWatch.ElapsedMilliseconds.ToString()); 

5. Implementation of calculations

5.1. Data structure declaration

Matrix A, vectors B and C, variables for storing their sizes will be defined globally:

 int[,] A; int[] B; int[] C; int m; int n; 

The TestFunction () method starts the calculation. First, in the method, the matrix is ​​multiplied by a vector using a sequential algorithm, then the same problem is solved using parallel computations. Consider this method.
Initializing data structures:

 nc = 2; //   m = 11000; //    n = 11000; //    A = new int[m, n]; B = new int[n]; C = new int[m]; 

Matrix A and vector B are generated:

 Random r = new Random(); for (int i = 0; i < m; i++) { for (int j = 0; j < n; j++) A[i, j] = r.Next(100); } for (int j = 0; j < n; j++) B[j] = r.Next(100); 

5.2. Sequential multiplication algorithm

Consider the SequentialMul method:

 Stopwatch sWatch = new Stopwatch(); sWatch.Start(); for (int i = 0; i < m; i++) { C[i] = 0; for (int j = 0; j < n; j++) C[i] += A[i, j] * B[j]; } sWatch.Stop(); Console.WriteLine("  = {0} .", sWatch.ElapsedMilliseconds.ToString()); 

5.3. Parallel multiplication algorithm

Parallel processing is performed by running multiple copies of the computational method. Each copy of the method performs processing of a certain part of the source data. To describe the task for each method, the InputData class is used:

 public class InputData { public int start; //   public int stop; //   } 

The class start / stop fields store the starting / ending line number of vector C, which is calculated using an instance of the computational method.
Consider the ParallelMul method:

 //       InputData[] ClArr = new InputData[nc]; for (int i = 0; i < nc; i++) ClArr[i] = new InputData 

Next, set the source data for each instance of the computational method:

 //       nc  int step = (Int32)(m / nc); //    int c = -1; for (int i = 0; i < nc; i++) { ClArr[i].start = c + 1; ClArr[i].stop = c + step; c = c + step; } 

A dispatcher is created with a pool of two threads:

 Dispatcher d = new Dispatcher(nc, "Test Pool"); DispatcherQueue dq = new DispatcherQueue("Test Queue", d); 

The port is described to which each instance of the Mul () method sends a message after the completion of the calculations:

 Port<int> p = new Port<int>(); 

The Arbiter.Activate method places two tasks in the dispatcher queue (two instances of the Mul method):

 for (int i = 0; i < nc; i++) Arbiter.Activate(dq, new Task<InputData, Port<int>>(ClArr[i], p, Mul)); 

The first parameter of the Arbiter.Activate method is the dispatcher's queue that will control the execution of the task, the second parameter is the task to be started.
Using the Arbiter.MultipleItemReceive method, a task (receiver) is started, which processes the receipt of two messages by the port p:

 Arbiter.Activate(Environment.TaskQueue, Arbiter.MultipleItemReceive(true, p, nc, delegate(int[] array) { Console.WriteLine(" "); })); 

The receiver is used to determine the end of the calculation. It will work only after two messages arrive at port p. In the delegate described in the receiver, you can include actions that must be performed after the completion of the multiplication process.
The Mul () method multiplies a part of a matrix by a vector:

 void Mul(InputData data, Port<int> resp) { Stopwatch sWatch = new Stopwatch(); sWatch.Start(); for (int i = data.start; i < data.stop; i++) { C[i] = 0; for (int j = 0; j < n; j++) C[i] += A[i, j] * B[j]; } sWatch.Stop(); Console.WriteLine(" № {0}:   = {1} .", Thread.CurrentThread.ManagedThreadId, sWatch.ElapsedMilliseconds.ToString()); resp.Post(1); } 

The Mul () method has two parameters:
1) an index that stores the value of the element of the array, which determines the parameters passed to the input method;
2) the port of completion to which the integer is sent after the completion of the calculations.
After completing the calculations, the Mul method sends an integer value to the port p (the sent value can be any).
The result of the calculations is shown in Fig. 3


Fig. 3. Results of calculations

Link to the source