rminnichAnother go round.776058d15 years agoHistory
309 lines · 18.0 KB · TeX
RawDownload
\subsection{Network programming models}Writing or converting applications to make best use of runtimes suchas MPI can be non-trivial: to avoid the synchronization imposed bymatching sends and receives, and barriers, and to overlapcommunication to reduce latency, programmers use non-blockingprimitives and asynchronous communication. In UNIX, a program can read and write on a file descriptor tocommunicate with a local device, or a TCP/IP port on a distantmachine, without changing the program text or even recompiling it: IOthrough file descriptors is location independent.  The intentionbehind developing specialized protocols---the bulk messaging protocolin particular---and the new messaging kernel interface is to allow HPCapplications to be written in a similar style, but with efficienttransport.   With IO through file descriptors, if two communicatingprocesses are on the same node, the system will use cross-domainreferences to transfer the data directly; if they are on differentnodes, the sender's data will be passed by reference through thekernel to the bulk messaging drivers, which will use DMA to transmitit to memory on the remote node, which will be passed by referencefrom the kernel to the receiving process.  File descriptors hide theprotocols used, not just the location.  When debugging on a laptop,pipes can replace the fancy networks, without changing the program, orthe communication semantics.  It is at least as easy to use as thenaive use of MPI's send/receive, but without imposing a static processstructure, since new processes can be created on demand, and added tothe communication set, in the same way as in network applications. The primitives are usable both directly by applications and bysupporting services.  For example, we ported to Plan 9 a smallscientific application that used MPI, in which asynchronouscommunication had resulted in a complex structure.  We replaced allMPI calls by calls to a small library using our primitives.  (Thelibrary provides \emph{reduce} and \emph{allreduce} operations, forinstance.)  The revised program is written in a straightforward style,like a simple network application, as if the communication weresynchronous, but the kernel and its protocol drivers exploitthe potential for concurrency and overlap.  As another example, wehave been experimenting with a small library and associated servicesthat creates a distributed spanning tree (which can be madefault-tolerant) to provide multicast and convergecast to supportlarge-scale computations. \subsection{Data network basics}The Blue Gene system provides several data networks.  The IO nodes arelinked by Ethernet, with connections through switches to the outsideworld, providing a conventional IP network.  The CPU nodes areconnected by the Torus; and the IO and CPU nodes are linked in acollective network.  The only path from a CPU node to the outside(including external file service) is through the collective to an IOnode, and then through the IP network on the Ethernet.  The Ethernetsupports large packets (`jumbo frames').  The collective and Torus,however, provide only small packets: 256 and 240 bytes of user datarespectively.  On BG/P, compared to BG/L, the Torus has extra DMAinterfaces to the Torus that can automatically fragment messages intothe network's tiny payloads, although the recipient must stillreassemble messages. The Plan 9 IP stack is completely independent of the MAC layers: asmall \emph{medium} driver connects the bottom of the IP stack to thefiles representing the driver for a given MAC level.  For example, theEther medium driver opens \texttt{/net/ether0/data}, reads and writesthe file, converting as it does so between the Ether packets thedevice expects and IP packets, adding and removing MAC headers, andinvoking ARP as required.  It was a simple task to write specificmedium drivers for each of the Torus and collective networks, justseveral hundred lines each.  Note that unlike some other systems onBlue Gene, Plan 9 therefore does not emulate an Ethernet on eithertorus or collective to implement IP. Not only is the MAC-level framingof IP packets specific to each device, but there is no need for ARP(which historically has been a problem when attempting to scale tothousands of nodes). Raw access (ie, non-IP) to each network is also provided, throughspecific device files in the name space, such as\texttt{/net/torusdata}; in fact, it is the same interface used by theIP medium drivers.  Applications that use the networks outside IPoften need to send messages larger than the hardware's payload.  Wetherefore added a simple fragmentation protocol, allowing messages ofup to 64 kbytes to be exchanged across the network.  Theimplementation is about 300 lines, allowing for an implementationentirely in software on BG/L, and a hybrid implementation on BG/P thattakes advantage on the Torus of support in the DMA subsystem forfragmentation, using a software header encoding that allows correctbut efficient reassembly at the receiver even when several nodes arecommunicating with the same recipient.  Note that combiningfragmentation with IP is worthwhile: the basic IP header is 20 bytes,and TCP adds 20 more, which is a non-trivial overhead for the tinyTorus and collective packets.  Fragmentation allows one TCP/IP headerto be used for a large message, which is then fragmented; althoughthere is overhead in fragment reassembly, the resulting messagetraverses the IP stack only once, instead of many packets. \subsection{Transport protocols} As noted elsewhere, Plan 9 uses the 9P protocol to implement a rangeof system and application services, including distributed services,such as those provided by the Unified Execution Model.  Unusually formodern network protocols, 9P is not 9P/IP: it is not expressly anInternet protocol.  It can use any underlying transport that providesreliable, sequential (in-order) delivery; the current version of 9Pdoes not even require the transport to respect 9P message boundaries.Having implemented an IP network above both Torus and collectivenetworks, and thus TCP/IP, we could immediately deploy 9P.  The BlueGene environment, however, like many supercomputer systems, makes iteasy to experiment with alternative protocol designs andimplementations.  Although the nodes, directly or indirectly, accessresources on outside networks such as file service or graphicsdisplay, much of the communication is intra- and inter-node.Furthermore, it is easy to change the protocol implementation at allnodes from run to run, simply as a by-product of rebooting the system.By contrast, there are many barriers to experimenting with newprotocols on the Internet.  For example, Plan 9 historically had areliable datagram protocol, called IL/IP, tuned to the needs of 9P.It has seen reducing use because on the wider Internet, interveningrouters and firewalls do not recognise the protocol; nor is it easy tochange them to do so. As well as supporting 9P, which is a reliable point-to-pointconversation between client and server, we wanted to support reliablemessaging between arbitrary pairs of processes (between any nodes).On Blue Gene, of course, the underlying hardware networks provide justsuch reliable delivery, at least in principle, and we had addedfragmentation to allow larger, more expressive messages.  On the otherhand, in the larger context of the project, we were interested infault-tolerance, including operation on unreliable or failingnetworks.  Furthermore, because of its intended use for a singleapplication using MPI, the hardware's properties are not quite rightfor 9P, or even general inter-process messaging.  Most obviously,reliability and flow-control are between nodes, not betweencollections of processes on those nodes.  Adding multiplexing ofinter-node channels, for example by running IP, requires keeping thereceiver `live' (ie, unblocked) at all times, to prevent the hardwareflow-control from triggering and causing deadlock. Thus, for reliablecommunication, the hardware level flow-control is neither necessarynor sufficient, and flow-control must be added somewhere in themultiplexing software stack. Multiplexing, datagram and transport protocols are superficiallystraightforward, but typically become burdened with much practicaldetail, for connection establishment, error control, flow control, andcongestion control. %\cite{John-Day-Patterns-of-Network-Archictures}Rather than risk the time to develop a complete protocol from scratch,we decided to adopt and adapt an existing complete protocoldefinition, or more probably a suitable subset of one, limiting theimplementation to just what we needed in the Blue Gene environment.We needed multiplexed messaging connections for 9P, and reliable butconnectionless messaging for computational work.  The first attemptwas loosely based on the Xpress Transport Protocol,\cite{xtp} whichhad the attraction of supporting unreliable and reliable datagrams,and reliable connections as supersets of each other.  The protocolessentially microcodes the effects desired by the application level;thus, a reliable datagram includes one protocol header that declares anew connection, transfers data into it, requests an acknowledgement,and closes the connection; a reliable connection instead puts thoseoperations in the headers of distinct messages, in particular leavingthe connection open after the first message, and taking advantage ofthe context thus established to abbreviate subsequent messages.  Plan9's TCP/IP has 3236 lines for the TCP level, 706 for the IP level, andsome auxiliary code.  By contrast, the XTP prototype had 1740 lines ofcode, and needed no IP level, but supported not just TCP-likeconnections, but reliable datagrams as well.  Unfortunately,connection establishment and reset in XTP still seemed overly complexfor this application, requiring message exchanges and timers.  Asecond attempt was made (reusing much of the code), inspired by a mucholder protocol \emph{delta-t}.\cite{watson1989delta} It was simpler than XTP,and had a unique, timer-based approach to connection management.Briefly, in delta-t a logical connection exists between every possiblesender and every possible receiver, but a connection isquiescent---thus requiring no state on either sender or receiver---ifno message has been sent or received for a specified interval.  Thedelta-t designers observed that timers are required to detect timeout, and to provide protection against lost messages on connectionclose, but the timers can be used instead to make the connectionmessages redundant.  Thus, to use a connection, a process simply sendsa message on it, and the receiver then creates the state for theconnection. After an optional acknowledgement, if no further messagesare received (or the receiving process has no reply), the connectionbecomes quiescent and the state is removed. In the original protocolthere are therefore no extra messages for connection establishment orshutdown, and the protocol can carry a single message efficiently.  Anold quip is that one cannot afford to have (say) TCP/IP connectionsbetween every pair of 65,000 nodes.  With the delta-t approach, thepair-wise connections (logically) exist but cost nothing, untilmessages are actually sent, when the transient state can carry as muchdata as needed, reliably, and without extra messaging for connectionestablishment. On the other hand, and currently the primary use, a delta-t connectioncan carry 9P traffic between a client and a 9P server. Because theinterface to the protocol (ie, \texttt{/net/pk}) obeys the Plan 9 namespace conventions for networks, applications can use the protocolinstead of TCP/IP without change. \subsection{Bulk messaging} On BG/L, the Torus network hardware presented a collection ofmemory-mapped FIFOs as its interface, accessed from the processor byexplicit loads and stores.  The BG/P Torus hardware hides that detailbehind a collection of rings of DMA descriptors, and new DMA hardwaremoves the data between memory and selected FIFOs.  The hardware goesmuch further, however, which required revisiting the software networkinterfaces to the Torus.  (The collective networks were unchanged.)In particular, the hardware provided new DMA operations: \emph{put}and \emph{remote get}, implemented by special packets interpreted bythe DMA engine.  The Put operation causes the hardware to send anarbitrarily-large block of memory from one node, across the Torus,directly into a block of memory on a specified target node.  Thedestination address is specified as an offset from a base address in atable on the remote node, selected by a `counter ID' specified by thesender.  In the other direction, one node sends a Remote Get packet toa second node, causing its DMA engine to execute a \emph{Put}operation.  Usually the target of the Put is the first node, so thatthe original message thus gets the relevant data from the second nodeefficiently using DMA. To exploit the BG/P extensions, the Plan 9 Torus driver was extendedto implement a lightweight protocol for bulk messaging. A processwishing to use bulk messaging opens the file \texttt{/net/torusbig},and sends and receives messages using write and read system calls. Thefirst 32 bytes of the data in each read or write is a software headergiving source and destination network addresses, an operation code, aparameter, and 24 bytes of uninterpreted data for use by theapplication.  That is followed by the data of the bulk message itself.The uninterpreted data might contain values for subaddressing, ormetadata describing the data, but that is up to the application. Theoperation code and its parameter are reserved to the protocolsubsystem. Bulk messaging is a network protocol. Writes and reads are notmatched, let alone synchronised, even in the implementation of theprotocol. A write causes a bulk message to be sent to the destinationnode, which is queued until some reader does a read.  The applicationdata attached to the message might be used to associate requests andreplies in an application level algorithm, perhaps by library code,but that is outside the messaging protocol. In the Torus driver, the protocol implementation uses the normal Toruspacket channels, but bulk message control packets are speciallymarked, and interpreted by the interrupt handler on the destinationnode. A bulk write request sends a Get control message to thedestination. The control message includes the metadata from therequest, and addressing information for the data. If the destinationcannot process the request, it will reply with a Nak. Otherwise, itwill issue a DMA Remote Get request, quoting the addressinginformation it was given. The hardware on both source and destinationnodes is programmed to signal an interrupt when the transfercompletes. The sending node marks the bulk request done; the receiverqueues the data to be read. Control packets carry tags to allowrequests to be cancelled, acknowledged, or nak'd if required. Note that the bulk writer is not required to wait for the transmissionto complete, allowing the usual overlap provided by buffering.  Moresubtly, although the transfer is logically a `put' from source todestination, the \emph{receiving} kernel initiates the transfer usinga DMA Remote Get.  There are several reasons. First, the parametersfor a DMA Put require the sender to specify the correct index in atable on the \emph{receiver}, and that can only be known by globalconvention or prior arrangement (ie, through a previous exchange ofmessages), neither of which supports dynamic communicationwell. Second, the receiver needs to find the memory to receive thedata, and might be receiving from many other nodes.  Having thereceiver start the transfer allows various forms of flow control.Most important, however, using DMA Remote Get is more robust againstnode failure.  If the sending node fails and stops, the request willfail or time out.  If the sending node fails and restarts, at worstthe Get request will fetch incorrect data, or receive an error. Themetadata can be used to check the data.  By contrast, if a DMA Put isused, and a destination node fails and restarts, it is possible that,without an error being notified, the put will transfer into whatevermemory is addressed at the time by the table index parameter,corrupting the target node. \subsection{IO segments} For initial experiment, we limited messages to 128 KiB, because ofaddressing constraints in the system. We did, however, want bulkmessaging to be able to send much larger chunks of data.  Furthermore,the normal read and write system calls imply copying, which isobviously inefficient.  We changed the system to allow bothrestrictions to be removed, by solving a more general problem. Weadded a simple form of cross-domain reference. A process can attach a special type of segment intended to be used forIO.  The segment is backed by physically-contiguous memory, usable byhardware DMA.  Inspired by the x-kernel \cite{hutchinson1991x}, fbufs\cite{fbufs}, and System 360's LOCATE mode, two new system calls wereadded:  \begin{center}      Message readmsg(int fd, uint iolimit)       int writemsg(int fd, Message m)\end{center} Like the read system call, readmsg fetches data from the file denoted by file descriptor fd,but instead of having the user provide a buffer address, readmsg returns a Message valuethat refers to the data read (ie, the buffer address), and any associated metadata (eg, source address). Similarly, writemsg accepts a Message value instead of a single buffer address.The data address in the Message can be an ordinary address in the process.but it can also be a reference to an IO segment.Such references can be passed between user process and device driver through the kernel.There are primitives to manage the memory in an IO segment. In the HARE implementation on BG/P, IO segments and the system calls are restricted to use with particular devices, specifically the Torus.Since then, however, we have revised and extended the scheme in the context of the 64-bit NIX work,to handle IO to arbitrary devices, and support interprocess communication.