Application-level dynamic scheduling of network communication for efficient re-partitioning of skewed data

Techniques are provided for using decentralized lock synchronization to increase network throughput. In an embodiment, a first computer sends, to a second computer comprising a lock, a request to acquire the lock. In response to receiving the lock acquisition request, the second computer detects whether the lock is available. If the lock is unavailable, then the second computer replies by sending a denial to the first computer. Otherwise, the second computer sends an exclusive grant of the lock to the first computer. While the first computer has acquired the lock, the first computer sends data to the second computer. Afterwards, the first computer sends a request to release the lock to the second computer. This completes one duty cycle of the lock, and the lock is again available for acquisition.

FIELD OF THE DISCLOSURE

This disclosure relates to computer network congestion. Techniques are presented for distributed dynamic scheduling of communication to avoid contention at network elements.

BACKGROUND

A complex, high-performance processor may have a core and coprocessors. The core offloads tasks to its coprocessors, which perform specialized functions for which they are optimized. The type and number of coprocessors depends on system aspects such as performance, the mix of tasks that need to be offloaded from the core, and power and size limits.

A challenge in developing scalable, distributed applications is efficient data exchange between participating computers of a distributed system. In large-scale systems, network performance and scalability are as critical as the performance of the computers. Some communication patterns such as all-to-all exchange are common within various application domains including high performance computing (HPC) and data analytics. However, these applications are traditionally difficult to optimize for network performance.

In the context of distributed query processing, some operations such as joins, aggregations, and sorts often need to repartition or otherwise redistribute the data across various computers in a system which may involve an all-to-all communication. The cost of data redistribution over a network may substantially increase query execution latency. As a result, efficient data redistribution is crucial for achieving high performance and scalability for distributed query processing. System throughput is further challenged by data skew, when computers have different amounts of data to send to each other during redistribution. Traditional techniques might work well for uniform data distribution, such as with shift-pattern communication, but perform poorly with data skew.

Some systems have non-blocking, high-bandwidth networks such as InfiniBand for reducing data communication time for an application. However, when the nodes perform all-to-all communication without any scheduling/ordering over an InfiniBand network, the bandwidth of the interconnect network is inefficiently utilized. As the number of communicating nodes in the system increase, the achievable network bandwidth may fall significantly below its peak.

The degradation in performance can be largely attributed to two reasons: 1) contention/congestion at a receiving endpoint, and 2) contention for a common inter-switch link in switches composed of a two-level fat tree. Contention at a receiving computer occurs when multiple senders simultaneously attempt to send data to a common receiver. As a result, all computers that attempt to send data to this common receiver may experience backpressure and degraded bandwidth.

Contention or oversubscription on inter-switch links occurs when communication between two or more independent pairs of computers makes use of the same inter-switch links which are used to connect the leaf and spine switches in a two-level fat-tree. Because each inter-switch link can only support the peak data rate between a single pair of computers, sharing of this link by more than a single pair leads to proportional reduction of bandwidth for each of the independent traffic flows that share a common link.

Furthermore, the above observations may be more or less applicable to bulk synchronous parallel (BSP) systems such as the shuffle phase of MapReduce. During a shuffle, each reducer pulls data from every mapper. Shuffle is another example of all-to-all data redistribution and may saturate switches. Data skew occurs when one reducer receives more data from mappers than another reducer receives during a same shuffle.

DETAILED DESCRIPTION

1.0 General Overview

3.0 Cross Talk

5.0 Fat Tree

8.0 Prioritization By Sender

10.0 Hardware Overview

1.0 General Overview

Techniques are provided for using decentralized lock synchronization to increase network throughput. In an embodiment, a first computer sends, to a second computer comprising a lock, a request to acquire the lock. In response to receiving the lock acquisition request, the second computer detects whether the lock is available. If the lock is unavailable, then the second computer replies by sending a denial to the first computer. Otherwise, the second computer sends an exclusive grant of the lock to the first computer. While the first computer has acquired the lock, the first computer sends data to the second computer. Afterwards, the first computer sends a request to release the lock to the second computer. This completes one duty cycle of the lock, and the lock is again available for acquisition.

In an embodiment with multiple switches interconnected by communication links, each inter-switch link is associated with its own lock, which may be hosted on any computer.

In an embodiment, denial of an acquisition request may bear information to assist a computer with deciding how to react to denial.

In an embodiment, an atomic network operation implements a remote lock. In an embodiment, additional behavior for acquiring multiple locks prevents deadlock.

2.0 Example System

FIG. 1is a scenario diagram that depicts interactions within an example system100, in an embodiment. Example system100uses lock synchronization to increase network throughput.

System100contains computers110,120, and (although not shown) at least one other computer. Each of these computers may be a rack server such as a blade, a personal computer, a mainframe, a smartphone, a networked appliance, or other networked computing device. Computers110and120communicate with each other over a computer network, such as a local area network (LAN) or internetwork of networks.

First computer110may have data to send to second computer120. However, a third computer may also have data to send to second computer120.

2.1 Network Contention

If the first and third computers were to send data to second computer120at the same time, their traffic would collide. For example, the network interface card (NIC) of second computer120may become saturated.

Alternatively, a shared communication resource such as a network element that is common to both transmissions could suffer contention. In either case, packet loss and retransmission may occur.

When two transmissions collide, one or both of them must be retransmitted, thereby reducing network throughput. System100uses lock synchronization to reduce collisions.

To reduce contention, the computers of system100serialize their data transmissions that are destined for a same recipient, such as second computer120. The computers of system100coordinate at traffic control to treat second computer120as an endpoint that can receive only one transmission at a time.

To achieve this coordination, second computer120contains lock130. Lock130may be used to manage mutually exclusive access to second computer120.

Lock130may be a hardware lock or a software lock. Lock130is mutually exclusive and may be a mutex.

However, lock130typically is not blocking. If lock130is unavailable (acquired by another computer), then a request to acquire lock130fails without waiting for lock130to become available.

Although not shown, first computer110may have its own lock to mediate inbound transmissions. For example, computers110and120may be peers in a distributed application, such as graph analysis, that is hosted on a Beowulf cluster.

As such, each computer of system100may have its own receiver lock, such as130. In an embodiment, the hosting of lock130is decoupled from the computer, such as120, for which lock130mediates access.

For example, lock130may be hosted on second computer120as shown, and yet be used by the computers of system100to mediate access to a different computer as a receiver, such as first computer110. This cross-hosting of locks may more evenly distribute network traffic by sending data to one computer and the data's lock control traffic to another computer.

In operation, first computer110may have application data to send to second computer120. A sequence of interactions occurs within system100to accomplish sending the data.

Interactions between computers110and120are shown as horizontal arrows, such as acquire141. As shown, the passage of time progresses downward. For example, interaction141occurs before142.

Having data to send to second computer120, first computer110attempts to acquire lock130by sending acquire141. Shown interactions, such as141, are communications such as a message or an invocation.

For example, a communication may be a hypertext transfer protocol (HTTP) request, a remote procedure call, an asynchronous message, a command such as an atomic hardware operation, or a network packet such as a datagram. Raw transport for a communication may be connectionless, such as user datagram protocol (UDP), or connection oriented such as transmission control protocol (TCP).

Second computer120receives acquire141and detects whether lock130is available or granted to another computer. This detection is shown as “lock available?”150.

With the exception of lock-control signaling, the computers of system100are configured not to transmit application data to second computer120without first acquiring lock130. If first computer110receives denial142, then first computer110must refrain from sending data to second computer120.

However, receipt of denial142does not mean that first computer110must block or sleep. Instead, first computer110may reattempt to acquire lock130by sending another acquisition request to second computer120.

If first computer110also has data to send to a third computer and receives denial142, then first computer110may attempt to acquire the lock of the third computer. If first computer110has already acquired that lock, then first computer110may send data to the third computer.

2.5 Acquisition Granted

Data144may be any bounded content stream, such as the contents of a file or buffer. By a convention such as an honor system or by actual enforcement, first computer110should refrain from holding lock130indefinitely.

2.6 Remote Lock

Each computer may host any number of locks, which are effectively remotely-accessible objects, whether inherently or by a container or other software wrapper. Hosting a lock by a computer includes the computer having ownership of data structures that are internal and/or private to the lock.

Lock hosting also includes exclusive responsibility for answering acquisition and release requests. Lock hosting further includes providing a remote interface to the lock to which other computers may send lock requests.

The remote interface may use various mechanisms to expose a lock. For example, a remote interface may accept data packets or network hardware operations. These implementation choices may determine, at a low level, what precise content is transmitted for the shown interactions.

In an embodiment, acquire141is transmitted as a single physical packet, such as an InfiniBand packet, an Ethernet packet, or a Fibre Channel frame. In an embodiment, grant143is transmitted as a single physical packet.

In an embodiment, acquire141may bear an atomic hardware operation that second computer120may process with little or no software execution. For example, the atomic operation may be a compare-and-swap (CAS), test-and-set (TAS), or fetch-and-add (FAA) such as provided by InfiniBand.

Lock implementation may be delegated to some objects or operations of an operating system that have suitable semantics. For example, a lock operation should not be idempotent.

In an embodiment, binding to a socket is not idempotent and constitutes granting a lock. In an embodiment, renaming a file is not idempotent and constitutes granting a lock.

2.7 Lock Release

For example, lock130may be leased to first computer110for a fixed duration or transmission of a fixed amount of data. In embodiments, the fixed amount of time or data is statically configurable or dynamically tunable.

In an embodiment, first computer110releases lock130immediately after data144is sent, even if the fixed duration has not elapsed, such as when a lease is still unexpired. In an embodiment, second computer120sends denials to other computers that request lock130while first computer110holds lock130.

First computer110releases lock130by sending release145to second computer120. Upon receipt of release145, second computer120may grant the next acquisition request that second computer120receives.

2.8 Starvation and Disassembly

First computer110may have more data to send to second computer120than can be sent during one lease (lock acquisition). For example, first computer110may need to send two megabytes, but a lock lease authorizes sending only one megabyte. Likewise, a lease may last one second, but sending two megabytes may take two seconds.

In these cases, first computer110may use a size threshold to disassemble a large data to be sent in chunks to second computer120. First computer110should reacquire and then release lock130for each transmission of a chunk.

This may help prevent starvation of other senders that need to send data to second computer120. Likewise, it may help prevent starvation of other receivers (besides second computer120) that first computer110has data ready for sending to.

3.0 Cross Talk

FIG. 2is a block diagram that depicts simultaneous data transmissions through a shared switch in example system200, in an embodiment. System200may be an embodiment of system100.

System200contains switch231that is connected by network communication links to computers211-212and221-222. Switch231may be a crossbar switch, a router, or other network element that can simultaneously relay different data streams to different recipients. However, switch231should not be a hub, because a hub does not simultaneously conduct multiple streams.

As shown and through switch231, computer211may send data to computer221at the same time that computer212sends data to computer222. For example, switch231may have multiple ports to enable concurrent transmissions.

However, concurrent transmissions require that no computer be a common endpoint for both transmissions. For example, contention occurs when both of computers211-212attempt simultaneous transmissions to computer221.

As such, a saturated endpoint (e.g. computer221) can cause contention. The locking techniques described above may reduce that contention.

4.0 Network Bridge

However, a saturated endpoint is not the only possible cause of contention.FIG. 3is a block diagram that depicts a saturated trunk that causes contention in example system300, in an embodiment. System300may be an embodiment of system100.

System300contains switches332-333and computers313-313and323-324. Switches332-333are connected to each other by a network trunk communication link.

For example, the trunk link may bridge separate LANs. For example, switch332and computers313-314may be components that occupy one LAN, while switch333and computers323-324may be components that occupy another LAN.

In an embodiment, each LAN and its components occupy a separate computer card, chip package, or system on a chip (SoC). In an embodiment, both LANs are virtual LANs (VLANs) that are backed by a same or different physical LANs.

Computer313may attempt to send data to computer323at the same time as computer314attempts to send data to computer324. As shown, both of these data transmissions pass through the network trunk.

Simultaneous attempts at transmission over the trunk may collide. If the trunk link is half duplex, such as a single wire, then a collision may occur regardless of whether both transmissions attempt to traverse the trunk in the same direction as shown or in opposite directions.

Unfortunately, the techniques described above for associating a transmission lock with each endpoint computer will not avoid the trunk contention shown. However, associating an additional lock with the trunk link itself can solve the trunk contention.

Ideally, any of computers313-314and323-324may host (contain) the trunk lock and mediate acquisition of the trunk lock by the computers. In an embodiment, either of switch332-333hosts the trunk lock. However, this is suboptimal because switches may be too busy to manage a lock.

The trunk lock should be acquired in conjunction with acquiring an endpoint lock. For example, computer313should acquire both of the trunk lock and the lock of computer323before sending data to computer323.

A network route between a sender and a receiver may traverse an arbitrary amount of switches and a corresponding amount of links that join pairs of those switches. Each involved link that joins two switches may have its own associated link lock.

For example, switches may be arranged together as an interconnection tree (hierarchical lattice), such that leaf switches (near the bottom of the tree) connect to many computers but few other switches, perhaps including only a single switch that is higher within the tree. Whereas other switches, such as root switches (near the top of the tree) connect mostly or solely to other switches.

A consequence of multi-lock transmissions is that each sender must know some or all of the network topology of the system in order to know which link locks to acquire. Indeed even in system302as shown, computer313should know that reaching computer323involves the trunk, whereas reaching computer314does not.

5.0 Fat Tree

FIG. 4is a block diagram that depicts example fat-tree system400, in an embodiment. System400may be an embodiment of system100.

System400switches, such as421-422and431, and computers such as411-413. The switches of system400are arranged in two layers (leaves and spines) as a fat tree. Each leaf switch, such as421, is connected to its computers, such as411-412, and some or all spine switches, such as431.

Each spine switch is connected to some or all leaf switches. For example, spine switch431is connected to multiple leaf switches, including421-422. Leaf switches need not be connected to each other.

Spine switches need not be connected to each other. An inter-switch link connects a leaf switch to a spine switch.

The tree is fat because inter-switch links are wider (higher bandwidth) than links between leaf switches and computers. For example, link441has more bandwidth than the link between leaf switch421and computer411.

The fat tree may be implemented as an InfiniBand topology. An InfiniBand network may employ a fat-tree routing algorithm.

Computer411may stream data to computer413over network links and through switches421-422and431. This route is shown with dashed lines.

Computers411-412may simultaneously try to stream data over link440, which is contentious. Contention between these two traffic streams is avoided by associating an additional lock with each fat (inter-switch) link.

For example, a lock may be associated with link440. The computers of system400should acquire the lock before sending data over link440.

To keep the overhead of managing the locks for the inter-switch links low, computers of the network host the inter-switch locks. When the number of inter-switch links is the same as the number of computers in a system, each computer hosts only one additional lock for an inter-switch link.

In an embodiment, an inter-switch link is wide because it is full duplex. For example, link442may actually be a pair of half-duplex links aligned in opposite directions.

Access to each half-duplex link may be mediated by a separate lock. For example, link442may actually have two locks.

In an optimistic embodiment, a sender need not acquire the lock of a downlink from a spine switch. For example, computer411may stream data up link441, through spine switch431, and down link442.

For that route, computer411should lock (acquire) link441but not442. As such, each sender should acquire only one inter-switch lock and one receiver lock to optimistically presume contention-free communication with another computer.

FIG. 5is a scenario diagram that depicts example interactions between computers within an example system500to acquire multiple locks, in an embodiment. System500may be an embodiment of system100.

System500contains computers511-512and521-522. Although not shown, computers521-522contain mutex locks for senders to acquire as a prerequisite of data transmission.

In this example, each of the computers of system500contains a portion of a distributed graph. At the beginning of a long-running analysis, each computer may start with an equal share of the graph.

However, accumulated graph mutations may eventually cause the allocation of graph data to become imbalanced (uneven). This may cause an overloaded computer to become a computational or communicative bottleneck that impacts the analysis throughput of system500.

Computer512may attempt to rebalance system500by redistributing (moving) some graph data from one computer to another. However, congestion is not the only problem that rebalancing may experience.

Coordination is further complicated by concurrent unrelated transactions. For example, client computer511may be simultaneously attempting to transfer graph data from sink computer521to source computer522, which is in the opposite direction of the graph data transfer by client computer512. For example, after grant545, possession of the locks of computers521-522is split between client computers511-512.

Client computer511holds the lock of sink computer521, while client computer512holds the lock of source computer522. However, both of client computers511-512still need to acquire another lock. For example, client computer512still needs to acquire the lock of sink computer521.

Each of client computers511-512holds one lock, still needs another lock, and neither computer can proceed. As such, system500is deadlocked.

However, such deadlock is avoidable by imposing heuristics for acquiring multiple locks. Each of client computers511-512should acquire all locks needed for a transaction before performing the transaction.

For example, client computer512should send both of acquires541and544and receive corresponding grants before transferring graph data between computers. Furthermore, all client computers should acquire locks in the same relative order.

For example, both of client computers511-512attempt graph data transfers that involve computers521-522. As such, client computers511-512should acquire the locks of computers521-522in the same order, regardless of which direction graph data is being moved.

For example, both of client computers511-512should acquire the lock of source computer522before acquiring the lock of sink computer521, even though client computers511-512are moving graph data in opposite directions.

When a sender holds one of two needed locks, the sender should release the held lock if the other lock is denied. For example, when client computer512receives deny546from sink computer521, client computer512should send release547to source computer522. Afterwards, client computer512may try again to acquire both locks in the same order.

6.4 Prioritization by Receiver

Rival acquires543-544may be contentious by more or less simultaneously arriving at sink computer521. Depending on the implementation, sink computer521may have discretion as to which of acquire543-544wins the requested lock.

For example, sink computer521may receive acquires543-544at different times, buffer both of them for a while, and then have to decide which of acquires543-544should get priority. As such, sink computer521may send a grant to one client and a deny to the other client.

In an embodiment, sink computer521generates a random number to decide which client gets the lock. In an embodiment, sink computer521grants its lock based on which of acquires543-544was first received.

In an embodiment, sink computer521grants its lock to whichever client has more data to send to sink computer521. For example, each of acquires543-544may indicate how much data does a client have to send to sink computer521.

In an embodiment, sink computer521grants its lock to whichever client has more data to send to all other computers combined. For example, each of acquires543-544may indicate how much data does a client have to send to all other computers combined.

6.5 Back Off

Within deny546, sink computer521may provide, and client computer512may consider, conditions that surround the denial. These conditions may indicate how much contention is the lock of sink computer521withstanding. Client computer512may analyze the conditions indicated by deny546to decide whether to immediately continue trying to acquire the lock or whether to defer a retry and perhaps negotiate data transmissions to another computer instead.

In an embodiment, deny546bears a count of denials sent by sink computer521since sink computer521last granted its lock. In an embodiment, deny546bears the sum of sizes of data transmissions denied by sink computer521since sink computer521last granted its lock.

Topological knowledge, route identification, lock acquisition, deadlock avoidance, and back-off may be concerns that are orthogonal to and apart from analysis application semantics. In an embodiment, some or all of these concerns are encapsulated into a software module or layer that is shared by, and more or less opaque to, applications.

In an embodiment, a distributed graph analysis framework encapsulates route and lock concerns. As such, the logic of a custom graph algorithm may ignore these concerns and yet benefit from them more or less effortlessly.

In an embodiment, route and lock concerns are encapsulated in a network driver. This is a generally applicable approach to support an ecosystem of unrelated applications.

An advantage of an encapsulation that is simultaneously shared by multiple applications is that contention can be reduced across a whole ecosystem of applications, instead of merely benefiting an isolated application that happens to embed the encapsulation. For example, this may achieve traffic control of multiple applications in a non-intrusive way.

8.0 Prioritization by Sender

FIG. 6is a block diagram that depicts an example system600that sends data from one computer to multiple computers, in an embodiment. System600may be an embodiment of system100.

System600contains at least computers610and621-622. Sender computer610has data that is ready to send to each of receiver computers621-622.

Receiver computers621-622may both receive data at the same time. However, sender computer610may send data to only one recipient at a time, such as when sender computer610has only one network interface card (NIC).

To avoid contention, sender computer610should serialize its transmissions to receiver computers621-622. For example, sender computer610may decide to send data to receiver computer621before sending data to receiver computer622.

In an embodiment, sender computer610generates a random number to select which of receiver computers621-622to send data to first. In an embodiment, sender computer610sends data to receiver computers in the same order that the data became ready to send, such as when sender computer610contains a queue of outbound data.

Sender computer610may have more data to send to receiver computer621than to receiver computer622. In an embodiment, sender computer610detects this imbalance and sends data to receiver computer621first.

FIG. 7is a scenario diagram that depicts example interactions between computers within an example system700to achieve pipelining, in an embodiment. System700may be an embodiment of system100.

For example, each of database computers721-722may store a partition of a database table. Although not shown, each of database computers721-722contains a mutex lock that a sender must acquire before sending.

In this example, client computer710has data to send to each of database computers721-722. For example, client computer710may access the table partition on each of database computers721-722. For example, client computer710may send a query to each of database computers721-722for processing against their partitions.

Client computer710may acquire the lock of database computer721, send data, and then release the lock. Afterward, client computer710may repeat this with database computer722.

However, system700need not fully serialize the data transmission transactions of client computer710. Client computer710may receive grant742from database computer721, at which time client computer710may begin transmitting data743, which may be a time consuming stream of many data packets.

Near the end of, or otherwise during, transmitting data743, client computer710sends acquire744to database computer722. In this way, client computer710achieves pipelining of transmission transactions, such that interactions of both transactions are interleaved.

10.0 Hardware Overview

For example,FIG. 8is a block diagram that illustrates a computer system800upon which an embodiment of the invention may be implemented. Computer system800includes a bus802or other communication mechanism for communicating information, and a hardware processor804coupled with bus802for processing information. Hardware processor804may be, for example, a general purpose microprocessor.

Computer system800also includes a main memory806, such as a random access memory (RAM) or other dynamic storage device, coupled to bus802for storing information and instructions to be executed by processor804. Main memory806also may be used for storing temporary variables or other intermediate information during execution of instructions to be executed by processor804. Such instructions, when stored in non-transitory storage media accessible to processor804, render computer system800into a special-purpose machine that is customized to perform the operations specified in the instructions.

Computer system800further includes a read only memory (ROM)808or other static storage device coupled to bus802for storing static information and instructions for processor804. A storage device86, such as a magnetic disk or optical disk, is provided and coupled to bus802for storing information and instructions.

The received code may be executed by processor804as it is received, and/or stored in storage device86, or other non-volatile storage for later execution.