Distributed Event Building

Distributed event building relates to the multilevel approach as multiprocessors to multicomputers [AG94]. In this scheme as developed by our group at CERN, readout units and filter units are no longer totally separated. Instead, each filter unit is connected by a SAN or bus to a readout unit. The architecture might even be extended to a single multibus computer (see figure 15 for architecture examples). Then no difference in hardware between the part that gets the data from the front end and the part that processes them is visible. Multiple buses are still necessary as the two units have to be able to operate independently to achieve the required performance. This combined data storage/processing unit shall be capable of performing filtering algorithms to the data that is held locally on demand. A filter unit that is member of such a cluster will not request data fragments anymore, but ask for results of calculations. It invokes a method of an event fragment proxy object locally. This call is forwarded with its parameters by the proxy object to the readout/filter unit combination that is responsible for the data that are needed to perform the computation. There processing of the data can be done locally. Only the results are returned to the caller. They serve as a basis for the initial requester to take or refuse this event. Only in case of acceptance have all fragments to be transferred to data store. This scheme of event building has several advantages:

• Reduction of data circulating in the switching network. Thus it exhibits a lower capacity requirement.
• Reduction of the switch size due to less data traffic and combination of readout/filter units.
• Higher parallelism, as fragments can be processed in parallel.
• No need for special purpose protocols to exploit maximal performance of the network. Ordinary communication libraries like BIP[PWT98,PT98] or MPI[For95] can be used without modification. Less coupling in the software is the result.
• Easier avoidance of hot spots in readout units. Some readout units, such as the ones that hold the muon detector data will be asked for data more frequently than others. This lies in the nature of the selection algorithms. As processing is done locally, those hot spots in data transfer are avoided. They are shifted to hot spots in processing, which can normally be solved more easily by stocking processing power.

There is however a significant disadvantage in this scheme: Some algorithms, such as reconstruction of a particle´s track, cannot be done with information from one readout unit only. The fundamental issue of massive parallel systems, i.e. how to exploit locality[JBD$^$91] has to be solved. Several segments of the detector have to be used to obtain a useful result. A possible approach is to request only results of calculations that are done in the readout units that hold the necessary data locally. If a queried readout unit needs more information, it invokes on its part calculation procedures on other nodes. Results could be cached locally, so that a readout unit that is asked twice for a certain calculation concerning the same fragment does not have to do further processing requests again.

How much data would have to be transferred in a simplified model of this approach? As algorithms for this kind of event building have not been developed yet, we can only give a rough estimation. Let us assume that information from the global trigger has to be requested in any case to start with processing. The fragment size is still 2 KBytes. Muon and Calorimeter information can be processed locally in each unit. Only the results (100 Bytes per message assumed) of a reduction to 75% of the trigger rate are transferred. For the other subdetectors the calculation is done as follows: The amount of data for sending the outcome of the locally applied algorithm is given. As these subdetectors cannot rely on local data only, we also add the amount of data that has to be transferred to them by other readout units to complete calculation. As the tracker is organised in layers, this will normally be done in steps. We assume that in each layer a computation for an element is requested not more than there are detector elements in the preceding layer. Suppose a track calculation is started, it can be continued in another readout unit, by just calling the function to calculate with the geometry parameters for the starting point of the next segment to which the algorithm interpolated (see figure 16). If the track is still existing in this unit, the result from the next segment that contains the continuation of the trajectory to sufficiently high probability is requested and so forth. As an estimate for the data to be transferred, we count again 100 Bytes for each readout unit that holds data of the involved subdetector. Computation is requested only once. The result stays available in the readout unit that requested the calculation and in the one that performed it until the fragment is cleared. Detector simulations used 93400 volumes[col98, p. 421]. Storing those in $\frac{100 {\rm ~KB} + 700 {\rm ~KB}}{2 {\rm ~KB}} = 400$ readout units, every unit will hold about 24 volumes. Let us furthermore assume that for every volume we will get some 4 Bytes of information as result of a calculation (the coordinates are known, so this one word per volume shall contain information about the hit). Then we can stay with the estimation mentioned earlier, of 100 Bytes per unit. As there are more than one volume per unit, we presume that every unit is only queried once. Then the result is returned to the requester and is available for other calculations as well. Only 400 messages are therefore estimated for the tracker per event: Each getting a result from each of the readout units that store information from the tracker subdetector. In this case we suppose that with the tracking detectors in each step, 50 percent more of the events can be rejected. The last step that cuts the rate down to 100 Hz is not shown as no more data transfer than the fully accepted event to data store is needed. The final cut can be made by combining the results of all requests either by voting or other kind of analysis. The amount to transfer the accepted event to offline storage is negligible.

We split the calculation into several terms. $c_1$ gives the data rate for the global trigger information, $c_2$ involves muon and calorimeters. $c_3$ affects the preshower, $c_4$ the tracker consisting of pixel detector and inner tracker.

$$\begin{eqnarray*} c_{1} = 200 \cdot 10 {\rm ~KB} = 1.953 {\rm ~MB/s} \end{eqnarray*}$$

The results from $c_2$ are retrieved from $30 = \frac{10 {\rm KB} + 50 {\rm KB}}{2 {\rm KB}}$ units. Each reply message is 100 Bytes large. The trigger rate is cut down to 75%.

$$\begin{eqnarray*} c_{2} = 200 \cdot 0.75 \cdot 30 \cdot 100 {\rm ~B} = 0.429 {\rm ~MB/s} \end{eqnarray*}$$

Preshower calculation now allows the rate to be cut down to 50%. However non local data is needed as well. The total size of 100 KB resides in $\frac{100 {\rm KB}}{2 {\rm KB}}$ units. Each of these units sends the result of its calculation to back the requester. Either the information needed can be calculated locally, is locally available already, or has to be retrieved by sending a request to another unit. As the volumes and thus the units can be grouped together according to their distance from the interaction point (i.e. in layers), not every unit will send messages to every other unit. For simplicity reason we assume now, that every readout unit sends its results only once. In case of sending the request two times, the required bandwidth doubles.

$$\begin{eqnarray*} c_{3} = 200 \cdot 0.75 \cdot 0.5 \cdot \frac{100 {\rm ~KB}}{2 {\rm ~KB}} \cdot 100 {\rm ~B} = 0.357 {\rm ~MB/s} \end{eqnarray*}$$

The same calculation is done for the whole tracker. Again the cut is 50%:

$$\begin{eqnarray*} c_{4} = 200 \cdot 0.75 \cdot 0.5 \cdot 0.5 \cdot \frac{800 {\rm ~KB}}{2 {\rm ~KB}} \cdot 100 {\rm ~B} = 1.43 {\rm ~MB/s} \end{eqnarray*}$$

$$\begin{eqnarray*} \sum_{i = 1}^{5} c_i = 4.1875 {\rm ~MB/s} \end{eqnarray*}$$

Suppose for the preshower and the tracking device each readout unit sends its result to more than one requester. One reason could be that it is not sure in which volume the track continues in a lower level. Let's assume that a calculation requires results from four underlying units. Then the tracker and preshower bandwidths are multiplied by four. The result is a total necessary bandwidth per port of about 10 MB/sec. This is still only a fraction of the initially proposed approach. Moreover, the switch size is reduced from $512 \times 512$ to at least $256 \times 256$. The results of the smaller switch are manifold:

• Reduction of cost.
• Simplification of the set up.
• Better exploitation of the switch performance as at least one readout unit and one filter unit share one switch port.
• Increased scalability as more processing units can be incorporated into one readout/filter unit.

However this scheme has not yet been investigated more detailed. Feasibility studies from subdetector groups are currently being carried out at CERN.

Although we shed light on the application domain that we have to serve, we did not yet capture its overall requirements on data acquisition. This, together with a selection of other experiments' systems and two significant advances in cluster computing that relate to our work will be presented in the next chapter.