>100 may identify multiple available devices for performing the requested operation. A device may be considered busy if the device is currently performing other operations and cannot be assigned further operations or is otherwise unavailable to perform graph processing operations. The device may be considered available if the device can be assigned further operations, e.g., the further operations can be queued for operation by the device.

The system 100 may partition the computational graph into multiple subgraphs. Each subgraph includes one or more nodes in the computational graph. The system 100 may, in some examples, obtain these subgraphs by breaking up pairs of nodes in the computational graph that are adjacent to each other, but are assigned to different devices.

The system 100 may assign, for each subgraph, the operations represented by the one or more nodes in the subgraph to a respective available device. In some implementations, the system 100 may assign each subgraph to a device having a computational capability necessary to perform the operations represented by the nodes in the subgraph. In some implementations, the request from the client includes data specified by a user that identifies a particular type of device to perform operations for particular nodes.

Thus, the system 100 generates an allocation of the computational graph across multiple devices that assigns each node in the computational graph to a respective device of the multiple devices. Each subgraph includes a given group of one or more nodes in the computational graph which, as described above, may be assigned to the same device. An exemplary subgraph to device mapping can be seen in FIG. 1B. In this example, exemplary subgraphs 140A and 140E of computational graph 140 may be allocated to GPU 116 and CPU 118, respectively. GPU 116 and CPU 118 may reside in a different or a same one of machines 126-132. In operation, GPU 116 and CPU 118 may perform the operations represented by the nodes included in the subgraph that the system 100 has assigned to them.

The system 100 may cause the devices to perform the operations of the nodes included in the subgraphs respectively assigned to the devices. In some implementations, the system 100 may send each device a request to start the operations of the nodes included in the subgraph assigned to the respective device. In response to receiving such a request from the system 100, each device may proceed to perform the operations of the nodes included in the subgraph assigned to the device.

Given that various dependencies may exist between nodes and groups of nodes in the computational graph, it follows that communication between various devices will need to be coordinated in order for such devices to be able to collectively execute the computational graph. For example, a node that represents an operation being executed on a first device may receive, as input, an output of another node that represents an operation being executed on a second, remotely located device. In this example, communication of the output of the operation represented by the other node from the second device to the first device may need to be coordinated in an effective manner in order to ensure proper and timely execution of the computational graph.

In some examples, the devices may coordinate exchanges of inputs and outputs with one another, as necessary. In these examples, the communication between the devices may occur independent of the system 100. That is, the system 100 may send each device a request to execute the operations of its respective subgraph and, in response to the request, each device may proceed to execute the operations of its respective subgraph in response to the system 100's request, which may include coordinating communication with other devices without additional intervention by the system 100.

In some implementations, to allow the devices to communicate independently of the system 100, the system 100 modifies the computational graph such that it includes additional nodes that represent communication operations between nodes. In particular, a device's respective subgraph may include a node representing an operation which, when executed by the device, allows the device to seamlessly communicate with another device that is executing a counterpart operation.

More particularly, the system 100 may insert “send” and “receive” nodes into the computational graph. Each send node represents an operation in which data, such as a tensor, is relayed to a receive node that is assigned to a different device than that of the send node. The system 100 may determine the locations in the graph at which to insert pairs of send and receive nodes by identifying any cross-device directed edges in the graph. Cross-device directed edges are directed edges in the computational graph that connect nodes in the computational graph that are to be allocated to two different devices. The system 100 may identify such cross-device directed edges while or after determining the allocation of the computational graph. Upon identifying each cross-device directed edge included in the graph, the system 100 may modify the computational graph by splitting each cross-device directed edge and inserting a pair of send and receive nodes in place of the cross-device directed edge. This modification process is described further below in reference to FIGS. 2A-C.

FIGS. 2A-C illustrate example computational graphs. By way of example, a computational graph system, e.g., the system 100 of FIG. 1A, can receive a request from a client to, given a set of inputs, compute an inference using the computational graph 200A. The computational graph 200A may receive input from an exemplary source 201 and provide output to an exemplary sink 215. The exemplary source 201 and sink 215 may, for example, be one or more other nodes of the computational graph.

The system may, for instance, determine to allocate computational graph 200A across three different devices. To make this determination the system may analyze the computational graph 200A to identify one or more nodes that can be partitioned into subgraphs and allocated to the devices available. For example, the system may determine to allocate a first chain formed by nodes 204 and 214 to a first device, a second chain formed by nodes 206, 208, and 210 to a second device, and node 202 to a third device. Although other possible chains of nodes are possible, the system can select the chains that minimize the number of subgraphs.

Under this particular allocation, the system may identify directed edges 220A and 230A as being cross-device directed edges. That is, the system may recognize that directed edges 220A and 230A each connect nodes that the system has determined to allocate to different devices. For example, it can be seen that directed edge 220A connects node 202 to node 208, which the system has determined to allocate to third and second devices, respectively. Similarly, directed edge 230A crosses device boundaries by extending from node 208, which corresponds to the second device, to node 212, which corresponds to the first device.

In response to identifying each cross-device directed edge included in computational graph 200A, the system may proceed to modify the computational graph 200A to generate a modified computational graph, e.g., the modified computational graph 200B illustrated in FIG. 2B. More particularly, the system may insert a pair of send and receive nodes between nodes 202 and 208 in place of directed edge 220A. In similar fashion, the system may also insert a pair of send and receive nodes between nodes 208 and 212 in place of directed edge 230A. It can be seen that directed edges are also inserted between each send or receive node and each node adjacent to such send or receive node.

Referring now to FIG. 2B, it can be seen that modified computational graph 200B, which represents a version of the computational graph 200A as modified by the system, includes a two pairs of send and receive nodes. Specifically, a first send node, S₁has been inserted between nodes 202 and 208, and a corresponding first receive node, R₁has been inserted between the first send node S₁and node 208. By these insertions, the system allocates the first send node S₁to the same device as node 202 (i.e., the third device) and allocates the first receive node R₁to the same device as node 208 (i.e., the second device). A bidirectional bridge 220B connects the first send node S₁to the first receive node R₁. The bridge 220B can be seen as representing a communication channel between the first send node S₁and the first receive node R₁.

When executed, the operations represented by the first send node S₁and the first receive node R₁enable the second and third devices to communicate in a predetermined manner. For example, execution of such operations may include the second and third devices adhering to a particular communication protocol. Address information of each respective device and/or node being executed by each respective device may also be built into these operations. In this way, the first send node S₁and the first receive node R₁may encapsulate every communication process that may need to be performed between the second and third devices in order for their respective subgraphs to be satisfactorily executed. These communication processes are described further below in reference to FIGS. 3 and 4A-B.

Similarly, a second send node, S₂has been inserted between nodes 208 and 212, and a corresponding second receive node, R₂has been inserted between the second send node S₂and node 212. By these insertions, the system allocates the second send node S₂to the same device as node 208 (i.e., the second device) and allocates the first receive node R₁to the same device as node 208 (i.e., the first device). A bidirectional bridge 230B connects the second send node S₂to the second receive node R₂, and can be can be seen as representing a communication channel between the second send node S₂and the second receive node R₂. Much like the first send node S₁and the first receive node R₁, the second send node S₂and the second receive node R₂enable communication to be easily coordinated between the second and first devices.

As demonstrated by the modified computational graph 200C in FIG. 2C, the system 100 may modify the allocation such that each send node is allocated to one respective subgraph and each receive node is allocated to another respective subgraph. For instance, the first send node S₁may be allocated to third device included in machine 130, along with node 202, as part of subgraph 240 that the system has assigned to the third device. Similarly, the first receive node R₁and second send node S₂may be allocated to the second device included in machine 126, along with nodes 206, 208, and 210, as part of subgraph 246 that the system has assigned to the second device. Furthermore, the second receive node R₂may be allocated to the first device included in machine 132, along with nodes 204, 212, 214, as part of subgraph 242 that the system has assigned to the first device.

At execution time, the operation represented by the first send node S₁may include a relaying of the output of node 202 to the first receive node R₁. This exchange may occur, for example, by way of channel 220B over a network 214 to which machines 130 and 126 have access. Network 214 may be similar to network 114 as described above in association with FIG. 1. The protocol leveraged to conduct this communication may be reflected in the operations represented by the first pair of send and receive nodes S₁and R₁, and may depend on one or more characteristics of the devices, machines, nodes, and networks associated with the execution of subgraphs 240 and 246. The operation represented by the first receive node R₁may, in turn, include a passing of the output of the operation represented by node 202 that it receives from the first send node S₁along as an input to node 208.

Similarly, the operation represented by the second send node S₂may include a relaying of the output of the operation represented by the node 208 to the second receive node R₂. This exchange may occur, for example, by way of channel 230B over the network 214 to which machines 132 and 126 have access. The protocol leveraged to conduct this communication may be reflected in the operations represented by the second pair of send and receive nodes S₂and R₂, and may depend on one or more characteristics of the devices, machines, nodes, and networks associated with the execution of subgraphs 246 and 242. The operation represented by the second receive node R₂may, in turn, include a passing of the output of the operation represented by the node 208 that it receives from the second send node S₂along as an input to node 212.

FIG. 3 is a flow diagram of an example process 300 provided through execution of exemplary send and receive nodes included in a computational graph. For convenience, the process 300 will be described as being performed by a system of one or more computers located in one or more locations. The computational graphs and operations described in association with process 300 may those which have been described above. For example, the process 300 may be performed through the execution of one or more portions of a computational graph as provided by system 100 as described above in association with FIG. 1.

As mentioned above, the operations represented by send and receive nodes may yield bidirectional communication between the devices to which they are assigned. Operations 330 and 340 represented by send and receive nodes S₃and R₃, respectively, may initially include a determination of a communication protocol to use for such bidirectional communication (332 and 342). The communication protocols leveraged by operations represented by pairs of send and receive nodes may depend on one or more characteristics of the devices, machines, nodes, and networks associated with the execution of the subgraphs at hand. This determination process is described further below in reference to FIGS. 3A-B.

The operations 330 represented by send node S₃may then act to determine whether output of an operation of upstream node 310 has been provided (310). Such output may include a tensor produced by way of execution of a subgraph that includes node 310 and send node S₃by an assigned device. Once provided, operations 330 represented by send node S₃may act to provide the output of the operation represented by node 310 as input to the operations 340 represented by receive node R₃, which may in turn act to provide the same as input to the operation represented by node 320.

Execution of operations 340 represented by receive node R₃may involve sending one or more messages to that of corresponding send node S₃(344). Such messages may serve as indication that the subgraph to which the receive node R₃belongs is ready to receive input by way of execution of corresponding send node S₃. In this way, these messages can be seen as a request to receive data output by one or more upstream operations. In the example of FIG. 3, the operations 340 represented by receive node R₃may receive input from send node S₃that includes the output of the operation represented by node 310.

At execution, the operations 330 represented by send node S₃may include a relaying of data in response to receipt of such messages. In some examples, the operations 330 represented by send node S₃may not act to relay the output of the operation represented by node 310 until such a message has been received (336). In this way, the flow of information between devices may be regulated so as to ensure that tensors are successfully exchanged. In these examples, the output of the operation represented by node 310 may be cached or otherwise stored in one or more memory regions local to the device that is executing the send node S₃(337). In some implementations, this output of the operation may be stored elsewhere on a machine to which the device belongs, or on one or more network-accessible devices. Once a receive node message is received, the stored output of the operation may be communicated to the party associated with receive node R₃(336 to 338) and subsequently flushed or otherwise deleted from memory. Upon determining that output from the operation represented by send node S₃has been received as input (346), the operations represented by receive node R₃may act to provide such output as input to the operation represented by downstream node 320 (348).

It can be seen that, through the execution of operations 330 and 340 as represented by send and receive nodes S₃and R₃, respectively, all data exchanges required for the execution of the associated subgraphs may be coordinated and carried out by the devices executing the subgraphs. The communicative autonomy provided to pairs of devices executing operations 330 and 340 may serve to shift the burden of coordinating communication away from the backend. In this way, a computational graph system, such as system 100, may be able to perform one or more other processes while devices are executing subgraphs, rather than actively regulate communication between such devices. For at least this reason, modifying computational graphs to include pairs of send and receive nodes may help reduce time costs and the amount of network communication required to process a computational graph in a distributed fashion.

In some implementations, operations 330 and 340 as represented by send and receive nodes S₃and R₃may be leveraged in handling all incoming and/or outgoing communications of a given device. For example, the operations 340 represented by receive node R₃may act to receive incoming communication from operations 330 represented by send node S₃, as well as from operations represented by one or more other send nodes. Similarly, the operations 330 represented by send node S₃may act to provide outgoing communications as input to operations 340 represented by receive node R₃, as well as to operations represented by one or more other receive nodes. In this way, operations 330 and 340 represented by send and receive nodes S₃and R₃may act to provide hub-like functions to the devices to which they are assigned.

Send and receive nodes S₃and R₃may serve to compartmentalize subgraphs in a manner that allows for a neural network or a portion of a neural network represented by such subgraphs to be trained on one device, and later on allocated to another device. Send and receive nodes S₃and R₃may also allow neural networks or portions of neural networks represented by one or more subgraphs to be trained or tested in new ways. For example, the operations 340 represented by receive node R₃may provide an input to the operation represented by node 320 that simulates a tensor being output by the operation represented by node 310 in order to train one or more portions of the computational graph downstream from the receive node R₃. In some implementations, a client device, such as client device 102 described above in reference to FIG. 1, may be leveraged to provide receive node R₃with such an input. In these implementations, the client device may execute an operation represented by a specialized node that provides a predetermined tensor as output to operations 330, which may also be executed on the client device. Such a specialized node can be seen as a substitute for nodes that are upstream in a computational graph from that of receive node R₃and node 320. In this way, users may be able to remotely test and/or train a neural network or a portion of a neural network represented by one or more downstream subgraphs by executing an operation represented by a specialized node and operations 330 on their client device. The tensor output by an operation represented by a specialized node of this type may, for example, be static, user-defined, randomly generated, or determined based on one or more characteristics of the devices, machines, nodes, and networks associated with the execution of the subgraphs at hand.

In some implementations, data exchanged between devices in association with send and receive nodes S₃and R₃may be compressed. That is, the operations 330 represented by send node S₃may act to perform one or more compression processes upon the output of the operation represented by node 310. Similarly, the operations 340 represented by receive node R₃may act to perform one or more decompression processes upon compressed data provided as output by way of execution of the operations 330 represented by send node S₃. The compression operations performed may include any conventional compression algorithm that is appropriate for transmitting data between the two devices. For example, the data exchanged between devices may be downconverted, truncated, or a combination thereof. Similarly, values conveyed by such data may also be subject to probabilistic rounding. Such compression operations may be selected based on one or more characteristics of the devices, machines, nodes, and networks associated with the execution of the subgraphs at hand. For example, compression operations may be selected based on the machine learning model's noise tolerance. Although compression has been described in association with operations 330 and 340, it is to be understood that such operations may leverage any of various signal processing and conditioning techniques.

As described above, the communication protocols leveraged by operations represented by pairs of send and receive nodes may depend on one or more characteristics of the devices, machines, nodes, and networks associated with the execution of the subgraphs at hand. FIGS. 4A-B depict two portions of computational graphs 400A and 400B that include send and receive nodes and are allocated to devices. It can be seen that the send node included in computational graph 400A has been assigned to device 412A, which in this example is a GPU, and that the receive node included in computational graph 400A has been assigned to device 414A, which in this example is also a GPU. In this example, GPU 412A and GPU 414A reside within a same machine 410A. Since devices that the send and receive nodes of computational graph 400A are both GPUs and both reside within a same machine 410A, it may be advantageous for their exchanges to be conducted under a remote procedure call (RPC) or other localized request/response protocol.

In some implementations, the particular communication protocol to be used may be determined through execution of the operations represented by send and receive nodes after subgraph allocation. Such determinations may be made, for instance, on the basis of address information known to these operations. In this example, GPU 412A and GPU 414A may execute operations associated with the send and receive nodes which indicate that the two devices both reside within machine 410A, and subsequently coordinate communications under RPC. In other implementations, a communication protocol may be predetermined and indicated within the operations represented by each send and receive node at the time of their insertion. In these implementations, each communication protocol may be determined in a manner similar to that which has been described above.

It can be seen that the send node included in computational graph 400B has been assigned to device 422B, which may be a CPU or GPU, and that the receive node included in computational graph 400B has been assigned to device 442B, which may also be a CPU or GPU. Device 422B may reside with machine 420B and device 442B may reside with machine 440B. A communication protocol may be determined for these devices through the use of techniques similar to those which have been described above. In this example, exchanges between devices 422B and 442B may be conducted under a remote direct memory access (RDMA) protocol. This protocol may be advantageous for devices of two different machines in that it permits communications to be conducted without involving the operating systems associated with each device. In this example, devices 422B and 442B may execute operations associated with the send and receive nodes which indicate that the two devices reside in different devices, and subsequently coordinate communications under RDMA.

FIG. 5 is a flow diagram of an example process 500 for modifying a computational graph to include send and receive nodes. For convenience, the process 500 will be described as being performed by a system of one or more computers located in one or more locations. For example, a computational graph system, e.g., the computational graph system 100 of FIG. 1A, appropriately programmed, can perform the process 500.

The system may obtain data representing a computational graph that includes multiple nodes and directed edge (502). For example, a computational graph may be obtained by a computational graph system after a receiving a request from one or more client devices. In some examples, each node in the computational graph is an instantiation of an operation.

The system may obtain data identifying an allocation of the computational graph across multiple devices (504). For example, a computational graph system may determine how each node included in the obtained computational graph is to be assigned across multiple available devices. In some implementations, this may include obtaining data identifying an allocation of the computational graph across hardware resources included in one or more machines.

The system may identify one or more directed edges in the computational graph that connect nodes that are assigned to different devices (506). For example, a computational graph system may identify one or more cross-device directed edges.

The system may insert send and receive nodes between pairs of nodes connected by the identified directed edges (508). For example, a computational graph system may replace each identified cross-device directed edge with a pair of send and receive nodes that are connected by a bidirectional bridge. In doing so, the system may further insert a directed edge between the send node and a respective first node, as well as insert a directed edge between the receive node and a respective second node.

In some implementations, data may be independently exchanged between operations represented by each node in a pair of send and receive nodes. In some instances, each node in a pair of send and receive nodes may represent an operation that receives, as input, an output of the operation represented by the other node in the pair communicated according to a protocol that corresponds to each hardware resource and machine to which the pair of send and receive nodes are assigned. As described above, such protocols may, for example, include RPC and RDMA.

Each send node may, for instance, represent an operation that receives, as input, an output of the operation represented by a respective first node, and provides the output of the operation represented by the respective first node, as output, to the receive node. The output of the operation represented by the respective first node may be a tensor, for example. Similarly, each receive node may represent an operation that receives, as input, the output of the operation represented by the send node, and provides the output of the operation represented by the send node, as output, to a respective second node. In some implementations, the operation represented by the receive node may further provide requests for output of the operation represented by the send node, as output, to the send node. In these implementations, the operation represented by the send node may provide the output of the operation represented by the respective first node, as output, to the receive node in response to one or more requests received from the receive node.

The system may modify the allocation to assign the send and receive nodes to devices (510). For example, a computational graph system may assign each send node to the same device as the node upstream from the send node, and likewise may assign each receive node to the same device as the node downstream from the receive node.

The system may assign operations represented by the nodes to the multiple device according to the modified allocation (512). For example, a computational graph system may request that each of multiple devices perform the operations included in a respective subgraph. In some implementations, the system may partition the computational graph into multiple subgraphs.

Embodiments of the subject matter and the functional operations described in this specification can be implemented in digital electronic circuitry, in tangibly-embodied computer software or firmware, in computer hardware, including the structures disclosed in this specification and their structural equivalents, or in combinations of one or more of them. Embodiments of the subject matter described in this specification can be implemented as one or more computer programs, i.e., one or more modules of computer program instructions encoded on a tangible non-transitory program carrier for execution by, or to control the operation of, data processing apparatus. Alternatively or in addition, the program instructions can be encoded on an artificially-generated propagated signal, e.g., a machine-generated electrical, optical, or electromagnetic signal, that is generated to encode information for transmission to suitable receiver apparatus for execution by a data processing apparatus. The computer storage medium can be a machine-readable storage device, a machine-readable storage substrate, a random or serial access memory device, or a combination of one or more of them. The computer storage medium is not, however, a propagated signal.