Big data transfer optimization

Embodiments are described for systems and methods that optimize large-scale data transfers over a wide area network by providing a data transmission protocol stack comprising a TCP layer that exchanges data processed by a host, and an IP layer that transports datagrams encapsulating the data to routers in the WAN, and a UDP-based transmission layer within the data transmission protocol stack that interfaces with the TCP layer and transmits data and control packets between the host and receivers of the WAN using a unicast duplex protocol. The stack has a WAN optimization components layer that interacts with the UDP-based transmission layer and provides transport protocol optimization through the UDP-based transmission layer and certain data de-duplication, compression, link aggregation, and application awareness functions.

TECHNICAL FIELD

Embodiments are generally directed to data storage systems, and more specifically to optimizing data transfers among distributed data centers.

BACKGROUND

Data migration between geographically distributed data centers is a critical task in modern large-scale computer networks. Due to growing amounts of transmitted data and limited throughput of channels traditional protocols like TCP (Transmission Control Protocol) are becoming outdated. TCP has proven to be very successful and greatly contributes to the popularity of today's Internet and still contributes the majority of the traffic on the Internet. However, TCP is not perfect and it is not designed for every specific application. In the last several years, with the rapid advance of optical networks and rich Internet applications, TCP has been found inefficient as the network bandwidth-delay product (BDP) increases. Though its AIMD (additive increase multiplicative decrease) algorithm reduces the TCP congestion window drastically, it fails to recover it to the available bandwidth quickly, and theoretical flow level analysis has actually shown that TCP becomes more vulnerable to packet loss as the BDP increases. Thus, the Internet transmission protocols must be optimized to maintain viability in heavy data traffic environments. Current methods of optimization however, often involve changing applications to accommodate different transmission protocols. This limits data mobility and imposes great cost overheads for system administrators.

FIG. 1illustrates present methods of transmitting data in large network systems using certain known transmission protocols. In system100ofFIG. 1, different networks102-106communicate to each other over a wide area network (WAN)110. Each individual network102-106may be a local area network (LAN) or other similar type of network comprising any number of server and/or client computers that are coupled together and then coupled to the WAN110through a gateway or network interface device. The networks102-106may represent data centers that include large-scale storage devices or storage networks and one or more servers to process the data to be stored and retrieved. In present systems, the standard communication of applications and data transfer among networks, such as102to106is performed with the TCP/IP protocol112. The TCP protocol generally does not run well on WANs when either latencies or packet drop rates are high, such as due to distance, bad network connections, congestion, and other similar factors.

Other protocols have been developed to overcome the deficiencies of standard TCP/IP, such as the BURST protocol from EMC Corporation. BURST is a replacement protocol for TCP that has proven to be reliable. It is built on top of the User Datagram Protocol (UDP) and is biased towards Big Data transfers, and was developed to overcome TCP's inefficiency in high bandwidth-delay product (BDP) networks with random losses. As shown inFIG. 1, this known alternative protocol114to TCP/IP112comprises the BURST layer on top of the UDP layer over the IP layer. The UDP layer is a connectionless protocol that emphasizes low-overhead operation and reduced latency in favor of error checking and delivery validation. Traditional approaches112that use TCP are generally not efficient enough to transmit big data volumes between datacenters, e.g.,102to106. The use of an alternative, more efficient protocol114, such as BURST often requires application changes, additional work and, in many cases may be infeasible to implement, such as if an application cannot be changed.

What is needed therefore, is a way to provide a transmission protocol without requiring changes in the applications so as to significantly improve data mobility, which is extremely important for big data stores synchronization and backup. Such a solution may be provided through the usage of standalone software module based on EWOC and implementing base TCP APIs (application programming interfaces) for invasive substitution of a standard operating system network modules.

DETAILED DESCRIPTION

Disclosed herein are methods and systems of optimizing big data transfers over large-scale networks without requiring changes to applications or undue system administration overhead. Embodiments extend existing known data transfer protocols by adding a wide-area network optimization layer to BURST and UDP layers.FIG. 2illustrates present methods of transmitting data in large network systems using WAN optimized transmission protocols, under some embodiments. In system200ofFIG. 2, different networks or data centers202-206communicate to each other over a wide area network (WAN)210. Each individual network202-206may be a local area network (LAN) or data center or other similar type of network comprising any number of server and/or client computers that are coupled together and then coupled to the WAN212through a gateway or network interface device. As inFIG. 1, the networks202-206may represent data centers that include large-scale storage devices or storage networks and one or more servers to process the data to be stored and retrieved.

Under some embodiments, the networks ofFIG. 2, e.g.,202and206communicate through an advanced TCP/IP protocol212that includes a BURST layer provided by EMC Corporation, which is an effective data transmission protocol over UDP, and standard TCP and IP layers. In accordance with the Open Systems Interconnection (OSI) standard, the TCP protocol is within the Transport Layer that performs host-to-host communications on either the same or different hosts and on either the local network or remote networks separated by routers. It provides a channel for the communication needs of applications. TCP provides flow-control, connection establishment, and reliable transmission of data. The IP protocol is within the Internet layer and has the task of exchanging datagrams across network boundaries. It provides a uniform networking interface that defines the addressing and routing structures used for the TCP/IP protocol suite, and its function in routing is to transport datagrams to the next IP router that has the connectivity to a network closer to the final data destination.

For the embodiment ofFIG. 2, the BURST layer is completely built on top of UDP as both data and control packets are transferred using UDP. It is a connection-oriented, message-oriented, unicast, and duplex protocol that optimizes bulk data transfers, such as replication. BURST preserves user message boundaries during network transfers. A user message is called the transaction in BURST, and BURST breaks the transaction down to data chunks. The size of the data chunks is fixed for any given transaction and does not exceed underlying transport MTU (UDP) including overhead. The last data chunk in transaction can be less than other chunks and is called the partial data chunk. Any given data chunk present on the receiver side is tracked in the bitmap. Each data chunk is associated with one bit in the bitmap and the bitmap holds all the bits for the transaction. The bitmap is divided into the ranges and the length of the range is based on number of bits that can fit into underlying transport MTU (maximum transmission unit), e.g., UDP, including overhead.FIG. 3illustrates the organization of BURST data for use in a WAN optimization system, under some embodiments. As shown in diagram300, the bitmap for BURST data in the network302comprises individual bitmaps with associated headers. These are then concatenated together to form the bitmaps in memory304, as shown. Individual data chunks are then organized into ranges to form the transaction data in memory306. The data chunks are then separated into individual units with associated headers to form the transaction data in the network308.

The BURST protocol comprises certain defined protocol data units (PDUs) that fit into the underlying transport layer MTU (UDP).FIG. 4is a table listing example packets used in the BURST protocol, under an embodiment. As shown in Table400, the Packet Type is a 4-bit field (bits0-3) that is presents in all BURST PDUs' headers and uniquely identifies its meaning, as shown in the Packet Type column of the table. It should be noted that the packet types and type values inFIG. 4are examples of certain packet type definitions, and other packet types are also possible. Certain PDUs are used to establish a connection, while others (e.g., DATA, RTT, BACKRTT and BITMAP) are used during the connected state, and still others are used during the disconnection procedure. Specific bit assignments, configurations, and definitions (e.g., flags, reserved bits, etc.) may be provided for each of the PDUs in accordance with the BURST protocol defined by EMC Corporation, or any other definition appropriate for specific implementations and network environments.

The appropriate PDUs are exchanged during relevant stages of the transmission process.FIG. 5illustrates the transmission of PDUs during the establishment, connected, and disconnection states of a BURST session, under an embodiment. Diagram500illustrates the message flows of PDUs among an initiator, listener and server for the main stages of connection establishment502, data connection (duplex data transfer)504and disconnection506for an example BURST session.

The BURST protocol defined herein is a UDP-based transport protocol that uses API semantics compatible with Berkeley sockets. It survives high latency and packet drops, and is optimized for best possible performance on links with high losses and delays while keeping memory consumption within given constraints. It employs smart flow control mechanisms optimized for best possible performance on large, medium, and small transfer sizes while keeping a low memory footprint. It is designed to be fair to other concurrent transfers. It employs smart available-link bandwidth probing mechanisms that allow the highest transfer speeds while releasing a fair share of bandwidth to other concurrent transfers in the network. It is designed and optimized primarily for big data transfers.

As shown inFIG. 2, the protocol stack for transfer212includes an EWOC layer over the BURST layer. In an embodiment, the EWOC layer comprises a WAN Optimization Components layer, which is a software package that uses optimized data de-duplication and compression algorithms along with BURST. The EWOC is configured to be used in two ways: (1) as a library that can be integrated into application, and (2) as an operating system-level feature when EWOC provides a network tunnel transparent for the application, though other implementations and configurations are also possible, and embodiments are not so limited. For this new software stack, the TCP functionality is provided on top of EWOC and BURST so that existing applications need not be changed to use more effective transport for the data exchange.

The software module and protocol stacks for the WAN optimized transmission212might be used in variety of ways to speed up the data exchange and augment volumes of transmitted data within the same physical channel. Embodiments may be applied in a number of different network environments. For example, a software defined data center, such as an elastic cloud storage (ECS) appliance solutions might benefit of the new transport protocol since it adds new quality of service without existing software changes. Another example is the iRODS (integrated Rule-Oriented Data System), which is an open-source, distributed data management software in use at research organizations and government agencies worldwide for creating data grids, digital libraries, persistent archives, etc. One example application is its use at genome research organizations, and other similar organizations. For this example use-case, replication and backup of genomic data is rather slow for traditional environment. Although iRODS is open source software and can be modified without approval of its owners the modification is generally not practical as modified software does not always have community support and can be difficult to implement and propagate. A solution not requiring application modification is thus much more preferable from maintenance point of view.

FIG. 6illustrates the use of the EWOC layer to transfer data among large, distant data centers, under some embodiments. As shown in system600, data centers602-606located in three major cities include a ViPR platform from EMC Corporation (or similar platform) along with data transmission interfaces including a respective EWOC interface. The ViPR platform represents storage automation software that provides a single automated way to abstract, pool and provision storage resources to deliver storage-as-a-service. Thus, the data centers602-606represent software-defined storage networks utilizing ViPR to manage and automate all storage resources for traditional and cloud storage platforms. Each data center includes at least one ViPR controller that comprises storage automation software that centralizes and transforms storage into a simple, extensible, and open platform. It abstracts and pools resources to deliver automated, policy-driven storage services on demand through a self-service catalog. Although embodiments are described with respect to a ViPR implementation, it should be noted that use of the EWOC interface is not so limited, and other implementations of data storage within data centers such as602-606may also be possible. System600illustrates an example in which large amounts of data may be transmitted among geographically distributed data centers. The data sets may be so large and/or complex as to constitute “Big Data” applications that cannot practically be processed using traditional data processing applications, such as relational database management systems (RDMS or DBMS).

System600shows a system built on a standard Linux kernel functionality, i.e., IP tables set up for transparent proxying (TProxy) and policy routing, used in conjunction with the EWOC-based process that transparently intercepts TCP transfers from different applications. In an embodiment, an EWOC-based application is called an EWOC Daemon (EWOCD) and, for instance, can be run on a physical or virtual Linux host in the same LAN segment. For a transmitting data center (e.g.,602) The payload from multiple TCP streams are aggregated, compressed and encapsulated into an EWOC protocol stack. The remote EWOCD (e.g.,606) does the reverse transformations, restores original TCP streams and distributes them to their final destinations. No modifications to existing applications are needed. The only changes are to the routing rules so that application traffic can be directed to the appropriate EWOCD home nodes. The optimization can be easily disabled to fall back to original implementations.FIG. 7is a flowchart that illustrates the above process steps (702to706) just described above. Thus, as shown inFIG. 7, the data transmitter aggregates, compresses and encapsulates the payload from multiple TCP streams into an EWOC protocol stack, block702. The receiver receives the transmitted data and disaggregates and decompresses the payload from the EWOC protocol stack, block704, and then restores the original TCP streams and distributes them to their final destinations, block706.

In an embodiment, the EWOC layer is a set of modular software components developed to provide building blocks for WAN transfer optimization tasks and provides the following functionalities: transport protocol optimization (e.g., BURST protocol as a TCP replacement); data de-duplication over the wire (e.g., ViPR W-EDRS); compression (e.g., ViPR C-EDRS); link aggregation; and application awareness. The EWOC Daemon application-aware service built on top of the EWOC stack and can run on any physical or virtual POSIX system and on multiple clients. It is used to deliver WAN optimization to systems where a built-in solution requires a lot of integration efforts and/or where optimization for multiple clients is required. The EWOC layer is built-on BURST, which is a highly optimized UDP-based transport protocol. The BURST protocol may be embedded in the EWOC layer, but it can also be delivered as a standalone product (e.g., software development kit).

FIG. 8is a block diagram that illustrates components of an EWOC library for use in a WAN optimized system, under some embodiments. A number of different data processing applications for transmitting, receiving, and/or processing transmitted or received data are organized into applications that are run on the same host802and applications that are run on other hosts804. The applications may be EWOC-aware or EWOC-unaware, as shown. Within the OSI model, the applications802and804operate within the application layer, in which the applications create user data and communicate this data to other applications on another or the same host. The applications make use of the services provided by the underlying, lower layers, especially the Transport Layer which provides pipes to other processes that are addressed via ports, which essentially represent services.

The applications802run on the same host as the EWOC library810are input directly to a facades interface802; while applications run on other hosts are input to the facades interface through a network connection (e.g., LAN806) and an IP table-based interception interface808. The facades interface receives the application data into appropriate client processes: TCP transparent and non-transparent clients, Unix domain sockets client and API client, as shown. A multiplexer combines the client data and inputs through a transformations component814that includes de-duplication, compression, and other appropriate processes for transmission through a transports component816. The transports component816transmits the data using the BURST, TCP, and other pluggable protocols from the EWOC library810to the WAN820. An application specific host interface component818provides an interface for user and system control over application operation.

As shown inFIG. 8, the transformations component814includes a compression function. This compression function may be a fast compression process using an LZ class algorithm, and that provides a choice of levels: LZ_L1 (faster); LZ_L2; or LZ_L3 (higher compression ratio). It may be a deep compression function using an LZH class algorithm that uses moderate CPU resources and provides a choice of levels: LZH_L1 (faster); LZH_L2; LZH_L3; or LZH_L4 (higher compression ratio). Other compression schemes may also be used.

The EWOC library810may be implemented through an application program interface (API). One receiving port is provided for all incoming connection requests and transfers to minimize UDP ports usage (RPNiPR). With respect to the interface with the BURST layer, the system multiplexes all EWOC pipes into one BURST connection to minimize overhead and contention. For the embodiment ofFIG. 8, the BURST protocol is illustrated as being implemented within the EWOC library, though it may also be provided as a separate functional component or software routine.

Embodiments of the WAN optimized transmission protocols and interfaces utilizing the EWOC library810and BURST protocol may be used in any appropriate data processing and storage environment, such as to transfer information to and from the cloud, process data for virtual machine images, provide data for data science analysis, transmit data from local hardware/devices, and provide data for streaming processing. Applications also include sharing the data in a hybrid cloud, and providing collaboration, backup, and Storage-as-a-Service (distributed storage) tools.

Embodiments may be applied to optimizing data transfers in practically any scale of physical, virtual or hybrid physical/virtual network, such as a very large-scale wide area network (WAN), metropolitan area network (MAN), or cloud based network system, however, those skilled in the art will appreciate that embodiments are not limited thereto, and may include smaller-scale networks, such as LANs (local area networks). Thus, aspects of the one or more embodiments described herein may be implemented on one or more computers executing software instructions, and the computers may be networked in a client-server arrangement or similar distributed computer network. The network may comprise any number of server and client computers and storage devices, along with virtual data centers (vCenters) including multiple virtual machines. The network provides connectivity to the various systems, components, and resources, and may be implemented using protocols such as Transmission Control Protocol (TCP) and/or Internet Protocol (IP), well known in the relevant arts. In a distributed network environment, the network may represent a cloud-based network environment in which applications, servers and data are maintained and provided through a centralized cloud-computing platform. It may also represent a multi-tenant network in which a server computer runs a single instance of a program serving multiple clients (tenants) in which the program is designed to virtually partition its data so that each client works with its own customized virtual application, with each VM representing virtual clients that may be supported by one or more servers within each VM, or other type of centralized network server.

The data generated and stored within the network may be stored in any number of persistent storage locations and devices, such as local client storage, server storage, or network storage. In an embodiment the network may be implemented to provide support for various storage architectures such as storage area network (SAN), Network-attached Storage (NAS), or Direct-attached Storage (DAS) that make use of large-scale network accessible storage devices, such as large capacity tape or drive (optical or magnetic) arrays, or flash memory devices.

For the sake of clarity, the processes and methods herein have been illustrated with a specific flow, but it should be understood that other sequences may be possible and that some may be performed in parallel, without departing from the spirit of the invention. Additionally, steps may be subdivided or combined. As disclosed herein, software written in accordance with the present invention may be stored in some form of computer-readable medium, such as memory or CD-ROM, or transmitted over a network, and executed by a processor. More than one computer may be used, such as by using multiple computers in a parallel or load-sharing arrangement or distributing tasks across multiple computers such that, as a whole, they perform the functions of the components identified herein; i.e., they take the place of a single computer. Various functions described above may be performed by a single process or groups of processes, on a single computer or distributed over several computers. Processes may invoke other processes to handle certain tasks. A single storage device may be used, or several may be used to take the place of a single storage device.

All references cited herein are intended to be incorporated by reference. While one or more implementations have been described by way of example and in terms of the specific embodiments, it is to be understood that one or more implementations are not limited to the disclosed embodiments. To the contrary, it is intended to cover various modifications and similar arrangements as would be apparent to those skilled in the art. Therefore, the scope of the appended claims should be accorded the broadest interpretation so as to encompass all such modifications and similar arrangements.