3. Processor & Connector
An intralogistic process consists of multiple individual processing steps which are linked to each other forming a sequence of steps. This library offers a set of classes which are used to model such a sequence of steps. Furthermore, the library distinguishes between processing steps, so-called Processors or Filters, and Connectors or Pipes for linking the processing steps. Both categories are represented by several classes – some of them are kept generic (i.e. abstract) such that individual requirements can be implemented.
A Processor (or also known as Filter) represents a single processing step where a working package (i.e. a Bundle) is processed. An instance of Processor (or its sub types) is commonly linked to a variable number of Workers which are executing the job of the Processor. Thus, a Processor can be seen as a workstation for a defined process step rather than as a process step itself, since the assigned Workers do the actual processing job. A Processor receives a clock signal, since as in a real assembly line, this clock signal triggers the execution and determines how fast a Bundle is processed. These clock signals are called ticks and the durations for processing are defined in ticks. A Bundle whose processing time has been set to 50 ticks, will therefore remain for 50 clock signal updates (i.e. 50 ticks) until the processing is completed and it is forwarded to the next step. The number of Bundles that are processed simultaneously depends on the implementation of the Processor, the implementation of the assigned Workers as well as the number of Workers.
In the following sections three types of supported Processors are introduced.
As the name implies, the Simple Processor is the simplest form of implementing a Processor. The actual processing step is not further specified, but delegated (i.e. out-sourced) directly to the assigned Workers. From a technical point of view, this means that the Simple Processor has a list of assigned Workers including methods for maintaining them. The clock signal is forwarded by calling the Update(long) method of each assigned Worker (in the same order as they are assigned in the list). Moreover, the Simple Processor has an input buffer for incoming Bundles waiting for processing. If a Worker is in idle state (e.g. it has just finished processing its last Bundle and has nothing to do), the Worker takes the longest waiting Bundle (first in first out principle) out of the input buffer for processing.
Example:
An instance of a Simple Processor can be used to simulate a scanning process. Each incoming Bundle represents a shipment containing 100 Entity objects which should be scanned. The scanning process takes 5 ticks per Entity respectively 500 ticks per shipment. Two Workers are assigned to this instance of Simple Processor; each Worker can process one shipment at the same time. At the beginning of the simulation, 7 shipments with 100 Entity objects each are placed in the input buffer. The following table shows the workload of the Workers at a specific time (in ticks).
| After X ticks | Number of Bundles processed by Worker 1 | Number of Bundles processed by Worker 2 | Number of Bundles in input Buffer |
|---|---|---|---|
| 1 | 1 | 1 | 1 |
| 501 | 1 | 1 | 3 |
| 1001 | 1 | 1 | 1 |
| 1501 | 1 | 0 | 0 |
| 2001 | 0 | 0 | 0 |
In comparison to the Simple Processor, this library provides also a more generic, abstract version of a Processor. Similar to the Simple Processor it contains a list of assigned Workers (including methods for maintaining them) as well as an input buffer for incoming Bundles, but the Update(long) method is unimplemented (i.e. abstract) such that individual requirements for a processing step can be implemented.
From a technical point of view, a Repository is neither a Processor nor a Connector. It has more in common with a Processor than a Connector. A Repository consists only of an input buffer without any mechanism for receiving clock signals nor managing any Workers. Bundles can be stored in this input buffer and taken out manually by another component/step.
Usage:
An instance of Repository can be used as a sink of a flow or as a clock-independent cache.
The implementation of the processing part encompasses two interfaces and tree classes which are explained in the following section:
IUnit:
The root type for all Processors as well as Connectors is IUnit. The IUnit interface defines the method In(Bundle) for inserting a Bundle object into the Processor (or Connector) and methods for setting/getting the Successor which is an instance of IUnit as well. This means that implemented instances of IUnit (e.g. a Simple Processor or a Router) can be linked to each other using the Successor property regardless of the sub type of the instance. Furthermore, a Bundle object is passed to a Processor or Connector by calling its In(Bundle) method.
IFilter:
The interface IFilter extends the defined methods of IUnit(and therefore extends IUnit) by defining further methods being relevant for the management of underlying Workers.
Processor:
The abstract class Processor implements the Worker management methods defined in IFilter as well as the methods for setting/getting the Successor and the In(Bundle) method defined in IUnit. In(Bundle) puts a Bundle object into the input buffer which is an attribute type of Queue. The size of the input buffer can be limited by setting InputBufferLimit. If this limit is exceeded at runtime, a InputBufferOverflowException is thrown. With Take() and Take(int) the input buffer can be dequeued by taking the longest waiting Bundle(s) out of the queue. The Workers are stored in a list being encapsulated by the Worker management methods implemented in this class and originally defined in IFilter.
Only the methods Update(long), Count() and EntityCount()are remain unimplemented (abstract) since this class is a generic representation and base class for any kind of processing steps.
SimpleProcessor:
SimpleProcessor inherits from Processor and implements the three methods Update(long), Count() and EntityCount(). As already described, SimpleProcessor is one of the simplest form of implementing a Processor. The clock signal which is received by calling the Update(long) method is directly forwarded to the stored Workers by calling their Update(long) methods (in the same order as the Workers are stored in the list). Count() returns the number of Bundles which are currently processed by all contained Workers and EntityCount() the sum of all underlying Entity objects stored in these processed Bundles.
Repository:
Repository implements the methods defined in IUnit but not those methods of IFilter or Processor since Repository does not have any Worker nor a mechanism for receiving a clock signal. Moreover, the methods for setting and getting a successor, which are defined in IUnit are not implemented (a NotImplementedException will be thrown if these methods are called), since a Repository has no successor. The class contains only an attribute for storing Bundles in a queue (FIFO) encapsulated by methods for inserting (In(Bundle)), taking and counting Bundles.
A Connector can be used to link two or multiple Processors to each other such that a sequence of steps can be formed. Since there are cases where a production process is not just a linear sequence of steps, the library provides different types of Connectors (i.e. not only a “straight-forward connector”), e.g. a Router having multiple targets. In comparison to a Processor, a Connector is not driven by a clock signal, but immediately forwards or modifies (depending on the Connector type) an incoming Bundle. Since Processors and Connectors are of the same root type, namely IUnit, they can be linked arbitrarily. Processors can be direct linked to each other as well as a Connector can be linked to one another Connector.
As a Processor can be directly linked to another Processor, there is no need for a dedicated Connector class implementing a straight-forward connection with one source and one target. Instead the target Processor is set as the successor of the source Processor (see IUnit.Sucessor(IUnit)).
A Splitter is a special Connector which splits an incoming Bundle into multiple sub Bundles by distributing the contained Entity objects of the incoming Bundle. The preferred size (number of Entites per Bundle) is specified. It is not necessary that the preferred size is a divider of the total number of Entities in the incoming Bundle without any remainder. If you specify 30 as the preferred size and there is an incoming Bundle having Entities for instance, then the Splitter will create four new Bundles: The first 30 Entities of the incoming Bundle will be distributed to the first Bundle, the next 30 Entities to the second Bundle and the following 30 Entities to the third Bundle. The remaining 10 Entities are distributed to the last Bundle which will be smaller than the other generated Bundles. Note that all header set in the incoming Bundle vanish after splitting.
A Merger is the counterpart to a Splitter. It buffers incoming Bundles and merges them to one new Bundle. The preferred size is specified. However, in comparison to the Splitter, it refers to the number of Bundles which should be merged instead the number of underlying Entities. Note that the Merger waits until the preferred number of Bundles has been received before a new Bundle is generated and forwarded. This might block the whole simulation and might result into a bottleneck. Furthermore, note that all header set in the incoming Bundles vanish after merging.
As already mentioned (see Bundle), headers can be added to a Bundle containing processing information or being relevant for routing decision. A Static Attribute Setter is a straight-forward Connector (i.e. connecting a source with a target) which adds a set of key-value pairs as headers to each Bundle flowing through this Connector. Existing headers will be overwritten.
Usage:
The Static Attribute Setter as well as the abstract Attribute Setter are used to set the duration (ticks) for the next processing steps or are followed by a Router which reads the set header and routes the Bundle depending on the header value.
The library includes an abstract version of the Static Attribute Setter where In(Bundle) is abstract and can be implemented based on your requirements. This allows to set individual header values depending on the analyzed characteristics of an incoming Bundle.
Example:
Consider the following scenario: Bundles should be routed to different assembly lines depending on the number of contained Entities. In(Bundle) checks the number of Entities for each incoming Bundle. If the number is lower than or equals 50, a header is added saying that the Bundle should be routed to Assembly Line A. If the number is higher than 50, a header is added saying that the Bundle should be routed to Assembly Line B. The following Router checks the header value and routes the Bundle to Assembly Line A or Assembly Line B.
A Router routes an incoming Bundle to one of its known destinations depending on the value of a specified header of the Bundle. Each route is a tuple of the value the header must have and the target being type of IUnit. Furthermore, a default route can be set which is applied if no other route matches the header value.
The implementation of the connecting part encompasses two interfaces and five classes. These five classes represent the previously presented Connectors.
