Non-native transactional support for distributed computing environments

Techniques are generally described for adding transactional support to a distributed storage environment. In various examples, first data may be written to a first set of locations in a distributed computer-readable non-transitory storage system through a non-transactional file system interface. In various further examples, metadata associated with the first data may be generated during the writing of the first data. In some examples, the metadata may be stored associated with the first data in at least a second location in a second computer-readable non-transitory memory. In some examples, a manifest may be generated defining a transactional commit of at least a portion of the first data. In some examples, the manifest may be generated by processing the metadata using first committer logic. In some further examples, the manifest may be stored in a third computer-readable non-transitory memory.

BACKGROUND

Computer networks often interconnect numerous computing systems to support their operations with the computing systems being co-located (e.g., as part of a local network) or located in distinct geographical locations and connected via one or more private or public intermediate networks. Data centers hosting large numbers of interconnected computing systems have become commonplace, such as private data centers that are operated by and on behalf of a single organization, and public data centers that are operated by entities as businesses to provide computing resources to customers. Some public data center operators provide network access, power, and secure installation facilities for hardware owned by various customers, while other public data center operators provide facilities that also include hardware resources made available for use by their customers.

DETAILED DESCRIPTION

Some computing service provider networks implement a variety of storage services, such as services that implement block-level devices (volumes). However, a number of applications running at data centers on a computing service provider network may face limitations with respect to use of some of the more common storage-related programmatic interfaces, such as various industry-standard file system interfaces. Some industry-standard file systems may have been designed without accommodation for the large-scale deployment of network-accessible services, and may therefore support consistency models and other semantics that are not straightforward to implement in distributed systems in which asynchronous computational interactions (e.g., asynchronous writes and/or reads by different compute nodes to/from a distributed network-accessible storage), failures of individual components, and network partitions and/or networking-related delays are relatively common. Additionally, various standard file system interfaces used in a distributed storage context lack transactional support and therefore may be unable to support features of ACID-compliant (atomicity, consistency, isolation, and durability) functionality.

Atomicity refers to an “all or nothing” aspect of a database transaction. If one part of the transaction fails, the entire transaction is treated as having failed and the database state is left unchanged. Atomic transactions ensure atomicity in situations such as power failures, errors and/or crashes. An atomic transaction appears to a device reading the committed data as an indivisible set of effects on a database. In various examples described herein, the databases (or other data structures) may be distributed among multiple different object block datastores. Consistency refers to a database property that brings the database from one valid state to another during a transaction. Transactions in consistent databases (or other data structures) cannot result in the violation of any defined rules for the database. Isolation refers to the property of a database that ensures that the concurrent execution of transactions results in the same system state as if the transactions were executed sequentially. Durability refers to the property of a database that ensures that once a transaction has been committed, it remains so, even in the event of power loss, system crash, or errors. For example, in a relational database, once a group of SQL statements execute, the results are stored permanently even if the database crashes immediately following execution.

Various embodiments of methods and apparatus for a high-availability, high-durability scalable distributed storage system that provides session management and configurable transaction commit logic to achieve transaction support over a standard file system interface (e.g., a non-transactional file system and an underlying object/block datastore) are generally described. In various examples, the systems and methods described herein may provide transaction support to distributed storage systems without modifying the base file system interface (and/or block data store interface) even where the base file system does not include native transactional support. Examples of non-transactional file systems (e.g., file systems without native transactional support) may include a standard UNIX file system. As used herein a transaction refers to an indivisible operation through which data that is the subject of a write operation is committed to a database as a single unit. During the transaction, a manifest (e.g., a data structure) of metadata representing data committed to the database(s) as a part of the committed transaction (e.g., describing data written as a transaction, according to the particular committer logic) is generated. The manifest is published (e.g., stored in a transaction store that is accessible by one or more compute nodes and/or clusters that are permitted to read the committed data). The storage of such a manifest renders the transaction visible to subsequent compute nodes and/or clusters of compute nodes such that the committed data may be read from the storage according to the manifest. Transactions either succeed or fail and may not partially complete (e.g., due to atomicity). Accordingly, transactions help ensure that data-oriented resources are not permanently updated in a data structure unless all operations within the transactional unit complete successfully. Transactions aid in error recovery (e.g., through a “rollback” of a transaction), and allow for consistent data state when a data structure (e.g., a database) is accessible by multiple client devices.

The various systems described herein allow individual compute nodes and/or clusters of compute nodes to write to a distributed object block datastore using a standard file system interface as though the object block data store is non-transactional. However, the various systems described herein may be effective to impart transactional support in a distributed computing environment using session management, user-configurable commit logic, and metadata generated during write operations. Additionally, the various committers, the metadata storage, session management, and the command logic storage are composable, such that a particular user may tailor the system to the user's individual needs for a particular system and/or task. Finally, from the perspective of a compute node (or cluster of compute nodes) reading data stored using the various systems described herein, a read appears to be a standard read operation as from a standard file system interface. Accordingly, in an example scenario using conventional file system read/write operations, a cluster of compute nodes reading a set of files during an on-going write operation (e.g., by a different cluster of compute nodes) may receive partially old files and partially new, overwritten files. In contrast, various transactional systems described herein may allow a read operation during an on-going write operation to provide the older version of the set of files until the write operation has completed and the committer has committed the new versions of the files to the object block datastore. During the commit a manifest describing the data written and/or including information as to how to read the data may be published so that various compute nodes and/or clusters of compute nodes may read data committed during the transaction. Additionally, rollbacks of corrupted data may be much simpler due to the transactionality imparted by the various systems described herein.

As used herein, a datastore may be a repository for storing and managing collections of data such as databases, files, emails, virtual machine images, etc. A datastore, as used herein, may generally refer to a storage architecture that manages data as blocks (e.g., fixed length chunks of data), files, and/or objects (e.g., a piece of data including a unique identifier). A distributed datastore may refer to a datastore that is distributed among multiple storage devices and/or systems. In various examples, distributed block datastores may be decentralized. A file system, as used herein, may refer to structure and logic rules for managing groups of data organized into files, as well as for managing the names of the files.

FIG. 1depicts a high-level overview of a distributed storage service, in accordance with various embodiments of the present disclosure. As shown, system100may, in some examples, be logically divided into various subsystems. For example, system100may comprise a storage subsystem130, a metadata subsystem120, one or more session managers110, one or more committers115, and/or one or more transaction datastores180. In various examples, storage subsystem130may comprise a plurality of nodes, such as storage nodes132a,132b, etc. Similarly, in various examples, metadata subsystem120may comprise metadata nodes122a,122b, etc. In various examples, the storage systems described herein may be distributed insofar as a plurality of different storage nodes132a,132b, etc., may be used to store data on one or more physical disks that, in turn, may be co-located and/or geographically diverse. In various examples, multiple compute nodes and/or clusters of compute nodes may have access to the distributed storage systems and may write to and/or read from the distributed storage systems as though the datastores were local to the compute nodes and/or clusters of compute nodes. A cluster of compute nodes may be a set of multiple compute nodes controlled by software to work together to perform one or more tasks. In various examples, a cluster of compute nodes may comprise a master compute node and one or more slave compute nodes. A master compute node may handle the scheduling and management of slave compute nodes.

Each compute node may, for example, be implemented as a set of processes or threads executed at a respective physical or virtualized server in some embodiments. The number of nodes in any given subsystem may be modified independently of the number of nodes in the other subsystems in at least some embodiments, thereby allowing deployment of additional resources as needed at any of the subsystems (as well as similarly independent reduction of resources at any of the subsystems).

In various examples, storage nodes132a,132b, etc., may represent some amount of storage space at some set of physical storage devices (which may be physically co-located or distributed at different geographical locations). In various examples, storage nodes132may store data objects, which may be organized as a set of logical blocks and each logical block may be mapped to a set of pages.

In the example depicted inFIG. 1, one or more compute nodes of cluster104amay be directed to write data to storage subsystem130. In the example, a master node of cluster104amay call and/or initialize session manager110at operation162. Session manager110may instantiate file system150, committer115, and metadata subsystem120(including metadata nodes122a,122b, etc.) at operation164.

One or more compute nodes of cluster104amay thereafter write data at operation166through file system150to one or more storage nodes132a,132b, etc., of storage subsystem130. While data is written by compute nodes of cluster104a, file system150may generate metadata associated with the data written at operation168. For example, as described in further detail below, the metadata may describe various statistics about the data written. The metadata may be sent to and stored at one or more metadata nodes122a,122b, etc., of metadata subsystem120.

After compute nodes of cluster104ahave concluded writing data to storage subsystem130, the master node of cluster104amay call committer115at operation170. At operation172, the committer115retrieves the metadata generated by file system150during the write operation166(e.g., metadata generated at operation168and sent to metadata subsystem120during write operation166). Committer115may be effective to use the metadata generated by file system150during the write operation166to generate a manifest of data committed as part of a transaction. Manifest may be sent to and stored by a transaction datastore180at operation174. Once the manifest is stored by transaction datastore180, or otherwise published, the write operation initiated at operation166is concluded. Although transaction datastore180is shown inFIG. 1as a distinct element, in various examples, transaction datastore180may be part of storage subsystem130and/or metadata subsystem120, etc. Additionally, transaction datastore180may be located wholly or in part at the cluster performing the write operation166. In various other examples, transaction datastore180may be located at one or more other locations apart from what is shown inFIG. 1.

As described in further detail below, committer115may generate a manifest comprising metadata describing the data committed as part the transaction. In various examples, the manifest may be a data structure that may, in effect, define the transaction for subsequent readers of the committed data. The particular committer logic of the committer115may generate the manifest such that the written data described in the manifest (and/or indications thereof) is different from the data written during operation166. For example, partial files may be combined into a single file and an indication of the single, combined file may be stored in the manifest. Similarly, temporary and/or “junk” files (and/or indications thereof) may be omitted from the manifest, such that the temporary and/or junk files are not considered part of the transaction, and as such may not thereafter be read by a reading node consulting the manifest. In general, the manifest defines a transactional commit of at least a portion of the underlying written data. The manifest describes the content of the transaction such that subsequent reading devices may retrieve and use the manifest to determine what and how to read data from storage subsystem130stored during the transaction. Committer115may send the manifest to transaction datastore180at operation174.

Subsequently, one or more compute nodes of cluster104bmay initiate a read from storage subsystem130. One or more compute nodes of cluster104bmay initiate a read request at operation176. The read request176may be a standard read request sent to file system150. At operation178, the file system150may retrieve the manifest stored at transaction datastore180. The manifest may provide service data, including a description of data, instructions for accessing the data, and/or a location of data stored during the transaction to file system150. Accordingly, file system150may return the requested data from storage subsystem130as in a standard read operation using file system150. Accordingly, using the various techniques described herein, transactional support may be imparted to a non-transactional file system in a distributed storage environment. Compute nodes and/or clusters of compute nodes reading from a distributed storage system (e.g., storage subsystem130) may reference the manifest of a transactional commit and may thereafter submit standard read requests to the file system. Accordingly, transactional functionality may be imparted without modifying the base file system interface of file system150.

In various examples, the metadata associated with data written to storage subsystem130may be stored locally on the compute node and/or cluster performing the write operation, while in various other examples the metadata associated with the files may be stored at one or more locations remote to the writing cluster or node. In other words, metadata subsystem120may be local to the writing cluster or may be located remotely (or some combination thereof). In at least some examples, the metadata associated with each file may be generated by file system150as a consequence of writing the file to the file system. Additionally, transaction datastore180may be local to the writing cluster or may be located remotely (or some combination thereof).

In various examples, the committer115may be effective to generate the manifest using the metadata stored at metadata subsystem120during the write operation (e.g., write operation166). The manifest may comprise a representation (in the form of metadata) of the files and/or other data written during the write operation. In at least some examples, the manifest may be registered (e.g., indexed) with the write location of the write operation so as to associate the manifest with the location of the written data. Further, in at least some examples, the manifest may include information describing how to access data written as part of the transaction.

Although metadata subsystem120and clusters104a,104bare depicted inFIG. 1as being separate logical components, in some examples, metadata subsystem120may comprise one or more components of a cluster or clusters performing a write operation. Additionally, although the example above describes writing files to the storage subsystem130, various other data types may be written (e.g., streams).

In various examples, session manager110may store metadata and/or may direct file system150to store metadata generated during the write operation166in metadata nodes122a,122b, . . . , of metadata subsystem120. Additionally, as previously described, committer115may generate a manifest identifying data written as a transaction. In some examples, the metadata and/or the manifest may be stored locally on the cluster performing the write operation, while in other examples, the metadata and/or the manifest may be stored remotely from the writing cluster (or stored at some combination of the local, writing cluster and remote locations).

A master compute node or other computing device may control one or more committers115to commit the data written as a transaction. Upon receiving a commit command170from the master node or other computing device, the committer115may retrieve the metadata generated during the write operation166from metadata subsystem120and may generate a manifest representative of the transaction and including detail regarding the data stored as part of the transaction and how to access the data stored during the transaction. Accordingly, transactional functionality may be introduced to a standard file system write operation in a distributed compute environment using metadata generated during the write operation and using committer logic to generate a manifest of the transaction.

In various examples, the metadata stored in metadata subsystem120may be stored in an ephemeral location and may be used by one or more committers115during a commit of the data to an object datastore. In an example where the metadata is stored in an ephemeral location, the metadata may be deleted from metadata subsystem120after the data is committed to the object datastore (e.g., after generation and storage/publication of the manifest). Advantageously, ephemerally storing the metadata may conserve computing resources such as memory and/or processing resources. However, in various other examples, the metadata generated during the write operation (e.g., write operation166) may be stored for longer periods of time and may persist in memory following the commit.

File system150may be a standard file system that may access storage subsystem130to perform writes of data and/or reads of stored data. The file system150may comprise a file system interface140through which a compute node(s) and/or a cluster may write data to and/or read data from storage subsystem130. In various examples, individual clients of the system100(e.g., one or more clusters and/or compute nodes) may supply their own commit logic pertaining to that particular client. As such, multiple committers115are depicted in system100ofFIG. 1. Accordingly, cluster104amay be associated with a first committer115and cluster104bmay be associated with a second committer115. Accordingly, the system100may enable non-transactional systems and file system-like block datastores to support transactions via user-defined commit logic in a distributed datastore. Additionally, such transactional functionality may be imparted without modifying the base file system interface.

In various examples, a transaction commit may be performed using a particular committer115. As previously described, in some examples, a particular committer115may be selected in various ways. In some examples, a particular committer115may be associated with a particular cluster104a,104b, or a particular compute node that is performing the write operation. In various other examples, the particular cluster and/or compute node that is performing the write operation may select a committer115by specifying the committer during the call to session manager110(e.g., operation162). The session manager110may thereafter instantiate the specified committer115. In at least some examples, the committer115may be associated with the file system150that the write operation166is using to write to the distributed storage system. For example, a High Density File System (HDFS) may be associated with an Apache Parquet specific committer. In various examples in accordance with the distributed storage service described herein, logic associated with a particular committer115may be effective to recognize and parse files of a given format (or formats) based on the structure of the file format. As described in further detail below, the committer115may be effective to commit files to a database in a transaction upon receiving a command from a cluster (e.g., commit command170) and/or from another computing device.

FIG. 2Adepicts an example of operations performed prior to a write operation, in accordance with various embodiments of the distributed storage service described herein.

In the example depicted inFIG. 2A, master node290of cluster214may call session manager110at operation201—Initialize Session Manager. In various examples, master node290may optionally specify a particular committer115in the call to session manager110. In various other examples, no committer may be specified and session manager110and/or file system150may select a committer115for a transaction. Session manager110may instantiate file system150, committer115, and metadata subsystem120. In various examples, file system150may be a non-transactional file system (e.g., a file system that does not natively support transactions). As previously described, metadata subsystem120may store metadata generated by file system150during a write operation (e.g., during the write operation depicted inFIG. 2B).

FIG. 2Bdepicts an example of operations performed during a write operation, in accordance with various embodiments of the distributed storage service described herein.

In the example depicted inFIG. 2B, cluster214, comprising a plurality of compute nodes, may perform write operation202to write files204to a file system of distributed storage subsystem130. Although, inFIG. 2B, files204are written, in various other examples other types of data may instead be written in accordance with the various embodiments described herein. File system150may write files204to storage subsystem130. As described in further detail below, at least some of the files204(and/or modified versions and/or portions thereof) may thereafter be committed as part of a transaction in accordance with the particular committer logic used to perform the commit. During the write operation of files204, file system150may generate metadata206associated with the files204. Session manager110may store (or may control file system150to store) metadata206in metadata subsystem120(including in, for example, metadata nodes122a,122b, etc.). As described in further detail below, metadata206may be used by committer115(not shown inFIG. 2B) to generate a manifest describing a transaction.

In various examples, metadata generated during the write operation (and associated with the data written therein) may describe characteristics of the particular files204written and may identify individual files from among other files. For example, metadata associated with a particular file may comprise an identification of encoding type, file format, tagging information, file pointers (and/or remote file locations if referencing locations outside the system), directory information, region identifiers, file length, file size, statistics about the files/data, filters, partition information, encryption information, etc.

FIG. 3Ais an example diagram describing a transaction commit320in a distributed storage service, in accordance with embodiments of the present disclosure.

After the completion of the write operation202depicted inFIG. 2B(e.g., after all files204, or other data have been written to storage subsystem130), master node290may send a commit command370to committer115. As previously described, committer115may be instantiated by session manager110(not shown inFIG. 3A). In various other examples, master node290may request a different committer115through session manager110after the completion of the write operation.

In response to the commit command370, committer115may retrieve the metadata (e.g., metadata206fromFIG. 2B) from metadata subsystem120during operation372. As previously described, the metadata may describe the data written to storage subsystem130during write operation202ofFIG. 2B.

Committer115may use the metadata retrieved at operation372to generate a manifest382describing data written during write operation202. In various examples, manifest382may represent data written during write operation202(and/or modifications thereof) committed as a transaction. The particular committer logic of the committer115may generate the manifest defining the transactional commit of at least a portion of the data written during write operation202. In various examples, the committer logic may modify data structures of the underlying written data, such that the data described in the manifest is different from the data written during the original write operation (e.g., write operation202). For example, committer115may determine that partial files written during the write operation202may be combined into a single file. Accordingly, committer115may store an indication of a single combined file (and/or instructions for reading the combined file as well as other information regarding the combined file) in manifest382. In another example, indications of temporary and/or “junk” files written during the write operation may be omitted from manifest382, etc.

In various examples, the commit logic may modify the original file structure during the commit (e.g., during generation of the manifest382). For example, during the write operation 20 part files may be written. Metadata generated during the write operation may represent the 20 part files. The particular committer logic of committer115may recognize (e.g., based on the file format structure of the 20 part files) the 20 part files as a single file and may represent the 20 part files as a single file of a different structure or format in manifest382. In various other examples, a write operation may commonly create a number of temporary files that are no longer deemed to be necessary and that are generated as a consequence of the write operation. Accordingly, the committer logic of committer115may be configured such that such temporary files are not represented in manifest382.

In general, the manifest382may describe (or define) the content of the transaction such that subsequent reading devices may retrieve and use the manifest382to determine what and how to read data from storage subsystem130stored during the transaction. Committer115may send the manifest382to transaction datastore180at operation374.

In various examples, committer logic may be user-defined to suit a particular user's purpose. For example, a committer may be designed to detect a particular directory structure of files written during write operation202using metadata retrieved at operation372. The committer may be designed to commit data based on the file directory structure to a few different partitions. For example, the file directory structure of the files may specify a region ID (e.g., United States, Mexico, Canada, etc.) and a marketplace ID (e.g., an identifier describing a particular region of a market, such as “North America”). A particular committer115may be selected that commits each separate region ID as its own transaction (e.g., each separate region ID may have its own manifest). Accordingly, in a first transaction all files specifying “United States” as the region ID may be committed in a first manifest, followed by a second transaction in which all files specifying “Mexico” are committed in a second manifest, and so on. In another example, a different committer115may be selected that commits all files with the marketplace ID “North America” as a single transaction (e.g., in a single manifest). Accordingly, users may select the committer115according to a desired implementation. In many examples, the file formatter may be associated with a committer115that is able to recognize and parse the way the particular file formatter writes the file format. Accordingly, each file format may be associated with a committer115or a set of committers115. As such, a user of the distributed storage systems described herein may not need to write their own unique committer115, but may instead use a default committer (e.g., default committer logic) for the particular file format in use.

FIG. 3Bis an example of operations performed during a read operation, in accordance with various embodiments of the distributed storage service described herein.

At operation376, one or more compute nodes of cluster214may send a read request376to file system150. In response, at operation378, the file system150may retrieve the manifest stored at transaction datastore180. The manifest may provide services, including a description of data and/or a location of data stored as part of a transaction. File system150may read the data specified in the manifest382according to read request376at read operation380. In an example, manifest382may omit so-called “junk” files written during the write operation. Accordingly, the read operation380may skip locations in memory corresponding to the junk files according to manifest382.

The read request376may be a standard read request sent to file system150. In response to the read request376, file system150may retrieve the manifest382and may read data according to the manifest382. Accordingly, file system150may return the requested data from storage subsystem130as in a standard read operation using file system150. Accordingly, using the various techniques described herein, transactional support may be imparted to a non-transactional file system in a distributed storage environment. Compute nodes and/or clusters of compute nodes reading from a distributed storage system (e.g., storage subsystem130) may submit standard read requests to the file system. Accordingly, transactional functionality may be imparted without modifying the base file system interface of file system150.

Advantageously, non-transactional file systems and file system-like block datastores may use the architecture described herein to support transactions via user-defined commit logic (and/or via default commit logic) without modifying the base file system interface. In a particular example, a distributed cluster of a plurality of compute nodes may be writing independently to a particular block(s) of data stored in a distributed storage system. However, because of the transactional functionality introduced using the various techniques described herein, readers reading the particular block(s) of data may all receive the same data. Essentially, the technology described herein allows the distributed storage system to behave as a non-distributed system for the purposes of accessing consistent data even while data is being continually written into object block datastores from one or more different nodes.

In another example, a single cluster may be designated to read a set of files. A different cluster may be designated to write the set of files. The cluster designated to write the set of files may begin overwriting the files with new data at a first time. Thereafter, before the completion of the write, the cluster designated to read the set of files may read from the files. In a non-transactional system, the reading cluster may receive half over-written files and half old files, or may get half deleted files and half written files that are corrupted, causing a system failure. Alternatively the writing cluster may crash before the write is complete leading to inconsistent data. The problem may be compounded when large amounts of data are written/read, as writes of terabytes or petabytes of data may take hours or even days, offering ample time for intermittent reads which may lead to corrupted and/or inconsistent data.

However, the various techniques and architectures described herein allow transactional functionality to be introduced within such systems without modifying the base file system interface. Accordingly, in the example above, the writing cluster will continue writing until completion and thereafter may commit the files in a transaction (e.g., by generating a manifest and publishing the manifest at the transaction store). Prior to the commit (e.g., prior to the publication of the manifest), the reading cluster will receive only the older version of the files. After the commit (e.g., after publication of the manifest), the reading cluster will receive only the new version of the files, eliminating the opportunity for data corruption. Additionally, rollbacks of data are made simpler, as an “un-commit” operation may be used to undo the commit, thereby rolling back to the older version of the data.

Referring toFIG. 4, the block diagram illustrates components of a computing device400, according to some example embodiments, able to read instructions424from a non-transitory computer-readable storage medium (e.g., a hard drive storage system or other memory) and perform any one or more of the methodologies discussed herein, in whole or in part. Specifically,FIG. 4shows the computing device400in the example form of a computer system within which the instructions424(e.g., software, a program, an application, an applet, an app, or other executable code) for causing the computing device400to perform any one or more of the methodologies discussed herein may be executed, in whole or in part. For example, the computing device may perform one or more of the functions of the data storage system described above with respect toFIGS. 1-3B. Computing device400may be an example of a compute node of a cluster of computing nodes (e.g., cluster104a,104bfromFIG. 1).

In alternative embodiments, the computing device400operates as a standalone device or may be connected (e.g., networked) to other computing devices. In a networked deployment, the computing device400may operate in the capacity of a server computing device or a client computing device in a server-client network environment, or as a peer computing device in a distributed (e.g., peer-to-peer) network environment. The computing device400may include hardware, software, or combinations thereof, and may, as example, be a server computer, a client computer, a personal computer (PC), a tablet computer, a laptop computer, a netbook, a cellular telephone, a smartphone, a set-top box (STB), a personal digital assistant (PDA), a web appliance, a network router, a network switch, a network bridge, or any computing device capable of executing the instructions424, sequentially or otherwise, that specify actions to be taken by that computing device. Further, while only a single computing device400is illustrated, the term “computing device” shall also be taken to include any collection of computing devices that individually or jointly execute the instructions424to perform all or part of any one or more of the methodologies discussed herein.

The computing device400includes a processor402(e.g., a central processing unit (CPU), a graphics processing unit (GPU), a digital signal processor (DSP), an application specific integrated circuit (ASIC), a radio-frequency integrated circuit (RFIC), or any suitable combination thereof), a main memory404, and a static memory406, which are configured to communicate with each other via a bus408. The processor402may contain microcircuits that are configurable, temporarily or permanently, by some or all of the instructions424such that the processor402is configurable to perform any one or more of the methodologies described herein, in whole or in part. For example, a set of one or more microcircuits of the processor402may be configurable to execute one or more modules (e.g., software modules) described herein.

The computing device400may further include a display component410. The display component410may comprise, for example, one or more devices such as cathode ray tubes (CRTs), liquid crystal display (LCD) screens, gas plasma-based flat panel displays, LCD projectors, or other types of display devices.

The computing device400may include one or more input devices412operable to receive inputs from a user. The input devices412can include, for example, a push button, touch pad, touch screen, wheel, joystick, keyboard, mouse, trackball, keypad, accelerometer, light gun, game controller, or any other such device or element whereby a user can provide inputs to the computing device400. These input devices412may be physically incorporated into the computing device400or operably coupled to the computing device400via wired or wireless interface. For computing devices with touchscreen displays, the input devices412can include a touch sensor that operates in conjunction with the display component410to permit users to interact with the image displayed by the display component410using touch inputs (e.g., with a finger or stylus).

The computing device400may also include at least one communication interface420, comprising one or more wireless components operable to communicate with one or more separate devices within a communication range of the particular wireless protocol. The wireless protocol can be any appropriate protocol used to enable devices to communicate wirelessly, such as Bluetooth, cellular, IEEE 802.11, or infrared communications protocols, such as an IrDA-compliant protocol. It should be understood that the communication interface420may also or alternatively comprise one or more wired communications interfaces for coupling and communicating with other devices.

The computing device400may also include a power supply428, such as, for example, a rechargeable battery operable to be recharged through conventional plug-in approaches or through other approaches, such as capacitive charging. Alternatively, the power supply428may comprise a power supply unit which converts AC power from the power grid to regulated DC power for the internal components of the device400.

The computing device400may also include a storage element416. The storage element416includes the machine-readable medium on which are stored the instructions424embodying any one or more of the methodologies or functions described herein. The instructions424may also reside, completely or at least partially, within the main memory404, within the processor402(e.g., within the processor's cache memory), or both, before or during execution thereof by the computing device400. The instructions424may also reside in the static memory406.

Accordingly, the main memory404and the processor402may also be considered machine-readable media (e.g., tangible and non-transitory machine-readable media). The instructions424may be transmitted or received over a network203via the communication interface420. For example, the communication interface420may communicate the instructions424using any one or more transfer protocols (e.g., HTTP).

The computing device400may be implemented as any of a number of electronic devices, such as a tablet computing device, a smartphone, a media player, a portable gaming device, a portable digital assistant, a laptop computer, or a desktop computer. In some example embodiments, the computing device400may have one or more additional input components (e.g., sensors or gauges) (not shown). Examples of such input components include an image input component (e.g., one or more cameras), an audio input component (e.g., a microphone), a direction input component (e.g., a compass), a location input component (e.g., a GPS receiver), an orientation component (e.g., a gyroscope), a motion detection component (e.g., one or more accelerometers), an altitude detection component (e.g., an altimeter), and a gas detection component (e.g., a gas sensor). Inputs harvested by any one or more of these input components may be accessible and available for use by any of the modules described herein.

As used herein, the term “memory” refers to a non-transitory machine-readable medium capable of storing data temporarily or permanently and may be taken to include, but not be limited to, random-access memory (RAM), read-only memory (ROM), buffer memory, flash memory, and cache memory. The machine-readable medium is non-transitory in that it does not embody a propagating signal. While the machine-readable medium is described in example embodiments as a single medium, the term “machine-readable medium” should be taken to include a single medium or multiple media (e.g., a centralized or distributed database, or associated caches and servers) able to store instructions424. The term “machine-readable medium” shall also be taken to include any medium, or combination of multiple media, that is capable of storing the instructions424for execution by the computing device400, such that the instructions424, when executed by one or more processors of the computing device400(e.g., processor402), cause the computing device400to perform any one or more of the methodologies described herein, in whole or in part. Accordingly, a “machine-readable medium” refers to a single storage apparatus or device as well as cloud-based storage systems or storage networks that include multiple storage apparatus or devices. The term “machine-readable medium” shall accordingly be taken to include, but not be limited to, one or more tangible (e.g., non-transitory) data repositories in the form of a solid-state memory, an optical medium, a magnetic medium, or any suitable combination thereof.

FIG. 5is a flowchart illustrating a first example process500for providing transactional support in a non-transactional distributed storage file system, in accordance with embodiments of the present disclosure. In some examples, the process ofFIG. 5may be performed by one or more other computing devices in a distributed computing network. For example, the process described inFIG. 5may be performed by one or more of the computing devices described below in reference toFIG. 7. The actions of the process flow500may represent a series of instructions comprising computer readable machine code executable by a processing unit of a computing device. In various examples, the computer readable machine codes may be comprised of instructions selected from a native instruction set of the computing device and/or an operating system of the computing device.

The process ofFIG. 5may begin at operation510, in which a compute node (e.g., master node290ofFIG. 2A) calls to a session manager routine (e.g., session manager110ofFIG. 1). In response, the session manager may be initialized and processing may proceed from operation510to operation520. At operation520, the session manager may instantiate a metadata subsystem (e.g., metadata subsystem120ofFIG. 1), a committer (e.g., committer115ofFIG. 1), and a file system (e.g., a non-transactional file system, such as file system150ofFIG. 1).

The process ofFIG. 5may continue from operation520to operation530aand operation530b. At operation530a, a compute node and/or multiple compute nodes of a cluster of compute nodes may begin writing a set of files to a non-transactional distributed storage file system. For example, cluster104amay write files through file system150to one or more storage nodes132a,132b, etc., of storage subsystem130as shown in operation166ofFIG. 1. During the write of the files, the file system instantiated by the session manager (e.g., the file system150ofFIG. 1) may generate metadata describing the files written to the distributed storage subsystem (e.g., storage subsystem130ofFIG. 1). As previously described, metadata associated with a particular file may comprise an identification of encoding type, file format, tagging information, file pointers (and/or remote file locations if referencing locations outside the system), directory information, region identifiers, file length, file size, statistics about the files/data, filters, partition information, encryption information, etc. At operation530b, metadata generated as a result of the write operation530amay be stored in the metadata subsystem instantiated by the session manager (e.g., metadata subsystem120). In various examples, the session manager may direct the file system to store the metadata in the metadata subsystem.

In the example provided inFIG. 5, a file system of the distributed storage system may not natively support transactionality. Additionally, although the process500describes writing files to the non-transaction distributed storage file system, the data written and committed as a transaction may not necessarily be files, but may instead be data of other types such as objects or blocks. In various examples, operations530aand530bmay occur in parallel, although in at least some other examples, the metadata generated as a consequence of the write operation530amay be stored in the metadata subsystem following the write operation.

Processing may continue from operations530aand530bto operation540, “Send commit command to committer”. At operation540, an instruction may be sent to the committer instantiated by the session manager (e.g., committer115) to commit at least a subset of the files written at write operation530aas a transaction. In various examples, the committer logic may be selected by the device issuing the commit instruction, while in other examples, the committer logic may be specified by a master compute node when the session manager is called. In various other examples, a default committer for the particular file format of the file system in use may be selected.

Processing may continue from operation540to operation550, “Retrieve, by the committer, metadata from the metadata subsystem”. At operation550, the committer logic may retrieve the metadata from the metadata subsystem instantiated by the session manager (e.g., metadata subsystem120) the metadata generated at operation530bdescribing the files written at operation530a.

Processing may continue from operation550to operation560, “Generate, by the committer, a manifest of the transaction.” At operation560, the commit logic may be effective to generate a manifest of the transaction (e.g., manifest382described above inFIG. 3A) based on the metadata retrieved from the metadata subsystem at operation550. The manifest may describe (or define) the data (e.g., the files and/or the modified files) committed as part of a transaction. The particular committer logic may be effective to process the retrieved metadata to generate the manifest. Processing the metadata may comprise transforming and/or aggregating the metadata as described herein. The particular committer logic may be effective to selectively omit and/or modify various data in the manifest. For example, if 5 part files are written during write operation530a, the committer logic may be effective to recognize that the 5 part files may be combined into a single file. Accordingly, the committer logic may provide instructions and/or service information in the manifest to describe a single, combined file committed during the transaction, as opposed to the 5 part files originally written. Additionally, in various examples, the particular committer logic may be effective to commit data according to various directories and/or partitions. Generally, the committer logic may define a transaction. However, the committer logic may be user specified and may be tailored to suit the particular implementation.

Processing may continue from operation560to operation570, “Store the manifest in transaction datastore”.

FIG. 6is a flowchart illustrating an example process600for providing transactional support in a non-transactional distributed storage file system, in accordance with embodiments of the present disclosure. In some examples, the process ofFIG. 6may be performed by one or more computing devices in a distributed computing network. For example, the process described inFIG. 6may be performed by one or more of the computing devices described below in reference toFIG. 7. The actions of the process flow600may represent a series of instructions comprising computer readable machine code executable by a processing unit of a computing device. In various examples, the computer readable machine codes may be comprised of instructions selected from a native instruction set of the computing device and/or an operating system of the computing device.

In various examples, although not depicted inFIG. 6, a compute node (e.g., master node290ofFIG. 2A) may initiate a call to a session manager routine (e.g., session manager110ofFIG. 1). In response, the session manager may be initialized and may instantiate a metadata subsystem (e.g., metadata subsystem120ofFIG. 1), a committer (e.g., committer115ofFIG. 1), and a file system (e.g., a non-transactional file system, such as file system150ofFIG. 1).

After instantiating the metadata subsystem, the committer, and the file system, the process depicted inFIG. 6may perform operations606aand606b. At operation606a, a compute node and/or multiple compute nodes of a cluster of compute nodes may begin writing first data to a first set of locations in a non-transactional distributed storage file system at a first time. For example, cluster104amay begin writing files through file system150to one or more storage nodes132a,132b, etc., of storage subsystem130as shown in operation166ofFIG. 1at a first time. In various examples, the first set of locations may refer to first locations in a data structure such as a distributed or non-distributed database. During the write of the data, the file system instantiated by the session manager (e.g., the file system150ofFIG. 1) may generate metadata describing the files written to the distributed storage subsystem (e.g., storage subsystem130ofFIG. 1). As previously described, metadata associated with a particular file may comprise an identification of encoding type, file format, tagging information, file pointers (and/or remote file locations if referencing locations outside the system), directory information, region identifiers, file length, file size, statistics about the files/data, filters, partition information, encryption information, etc. At operation606b, metadata generated as a result of the write operation606amay be stored in the metadata subsystem instantiated by the session manager (e.g., metadata subsystem120). In various examples, the session manager may direct the file system to store the metadata in the metadata subsystem.

The process ofFIG. 6may continue from operations606a,606bto operation608, “Receiving, at a second time after the first time, a first read request to read from the first set of locations.” At operation608, the file system instantiated by the session manager (e.g., file system150) may receive a first read request to read from the first set of locations. In response to the read request, the file system may retrieve a previously-stored manifest to determine the appropriate data to be read in response to the read request. In the example, the write operation initiated at operation606amay not yet have completed by the second time at which the first read request is received. The first read request may be generated by a different compute node and/or a different cluster of compute nodes relative to the compute node or nodes initiating the write of first data at operation606a. In various examples, the first read request may be a request to read data stored at a particular location or locations in a data structure (e.g., within a particular field or fields of a distributed database). Accordingly, the first read request may result in use of a previously-stored manifest to read data stored as part of a previous transaction.

The process ofFIG. 6may proceed from operation608to operation610, “Returning second data stored at the first set of locations in response to the first read request.” At operation610, the access subsystem may return data stored at the first location to the compute node and/or cluster issuing the first read request. In various examples, the data returned may represent data stored at the first set of locations prior to the commit of the first data currently being written to the first set of locations.

Processing may continue from operation610to operation612, “Receiving, at a third time after the second time, a commit command to commit at least a portion of the first data using first committer logic”. At operation612, an instruction may be received by the committer instantiated by the session manager (e.g., committer115) to commit at least a subset of the first data written at write operation606aas a transaction. In various examples, the committer logic may be selected by the device issuing the commit command, while in other examples, the committer logic may be specified by a master compute node when the session manager is called. In various other examples, a default committer for the particular file format of the file system in use may be selected.

Processing may continue from operation612to operation614, “Retrieving, by the committer, metadata from the metadata subsystem.” At operation614, the committer logic may retrieve the metadata from the metadata subsystem instantiated by the session manager (e.g., metadata subsystem120) the metadata generated at operation606bdescribing the first data written at operation606a.

Processing may continue from operation614to operation616, “Generating, by the committer, a manifest of the transaction and storing the manifest.” At operation616, the commit logic may be effective to generate a manifest of the transaction (e.g., manifest382described above inFIG. 3A) based on the metadata retrieved from the metadata subsystem at operation614. The manifest may describe (or define) the data (e.g., the files and/or the modified files) committed as part of a transaction. The particular committer logic may be effective to selectively omit and/or modify various data in the manifest. For example, if 5 part files are written during write operation606a, the committer logic may be effective to recognize that the 5 part files may be combined into a single file. Accordingly, the committer logic may provide instructions and/or service information in the manifest to describe a single, combined file committed during the transaction, as opposed to the 5 part files originally written. Additionally, in various examples, the particular committer logic may be effective to commit data according to various directories and/or partitions. Generally, the committer logic may define a transaction. However, the committer logic may be user specified and may be tailored to suit the particular implementation. As previously described herein, the manifest may be “published” (e.g., stored) at a location in one or more memories (e.g., transaction datastore180) that is available to properly authenticated compute nodes and/or clusters of compute nodes so that the authenticated compute nodes and/or clusters can retrieve the manifest data describing the transaction.

Processing may continue from operation616to operation618, “Retrieving the manifest.” At operation618, a compute node and/or cluster of compute nodes preparing to request a read of the transactional data may retrieve the manifest (e.g., from transaction datastore180). The retrieved manifest may provide descriptions, locations, and/or services related to data committed as part of a transaction by the committer, in response to the commit command sent at operation612. The manifest retrieved at operation618may be the manifest stored at operation616and may represent a transactional commit of at least a portion of the first data written to the first set of locations at operation606a.

Processing may continue from operation618to operation620, “Receiving, at a fourth time after the third time, a second read request to read from the first set of locations.” At operation620, a second read request may be received by the file system (e.g., file system150). The second read request may be generated by a different compute node and/or a different cluster of compute nodes relative to the node and/or cluster initiating write operation606a. Additionally, the second read request may be generated by a different compute node and/or a different cluster of compute nodes relative to the first read request of operation608. In various examples, the compute node and/or cluster of compute nodes may use the manifest retrieved at operation618to specify the data to be read. The manifest describes the first data written during operation606aas committed during a transaction according to the particular committer logic. Accordingly, the manifest allows a reading device to determine what data was committed during a transaction and how to access that data.

Processing may continue from operation620to operation622, “Returning the portion of the first data stored at the first set of locations in response to the second read request.” At operation622, the file system may return the data requested at operation620. In various examples, the portion of the first data committed during the transaction (as represented by the manifest) may, in some cases, be modified relative to the first data written at operation606aaccording to the particular committer logic used to generate the manifest. Accordingly, the data returned at operation622may not directly correspond to the first data written at operation606a.

An example system for sending and providing data will now be described in detail. In particular,FIG. 7illustrates an example computing environment in which the embodiments described herein may be implemented.FIG. 7is a diagram schematically illustrating an example of a data center75that can provide computing resources to users70aand70b(which may be referred herein singularly as user70or in the plural as users70) via user computers or other network-connected devices72aand72b(which may be referred herein singularly as computer72or in the plural as computers72) via network203. In various examples, compute nodes described in reference toFIG. 1may be an example of a computer or other network-connected device72aand/or72b. Additionally, in some examples, the network-connected devices72a,72bmay be configured in communication as a part of a cluster of compute nodes, such as clusters104a,104bdescribed inFIG. 1. Data center75may be configured to provide computing resources for executing applications on a permanent or an as-needed basis. The computing resources provided by data center75may include various types of resources, such as gateway resources, load balancing resources, routing resources, networking resources, computing resources, volatile and non-volatile memory resources, content delivery resources, data processing resources, data storage resources, data communication resources and the like. Each type of computing resource may be available in a number of specific configurations. For example, data processing resources may be available as virtual machine instances that may be configured to provide various web services. In addition, combinations of resources may be made available via a network and may be configured as one or more web services. The instances may be configured to execute applications, including web services, such as application services, media services, database services, processing services, gateway services, storage services, routing services, security services, encryption services, load balancing services, application services and the like.

These services may be configurable with set or custom applications and may be configurable in size, execution, cost, latency, type, duration, accessibility and in any other dimension. These web services may be configured as available infrastructure for one or more clients and can include one or more applications configured as a platform or as software for one or more clients. These web services may be made available via one or more communications protocols. These communications protocols may include, for example, hypertext transfer protocol (HTTP) or non-HTTP protocols. These communications protocols may also include, for example, more reliable transport layer protocols, such as transmission control protocol (TCP), and less reliable transport layer protocols, such as user datagram protocol (UDP). Data storage resources may include file storage devices, block storage devices and the like.

Data center75may include servers76aand76b(which may be referred herein singularly as server76or in the plural as servers76) that provide computing resources. These resources may be available as bare metal resources or as virtual machine instances78a-d(which may be referred herein singularly as virtual machine instance78or in the plural as virtual machine instances78). Virtual machine instances78cand78dare rendition switching virtual machine (“RSVM”) instances. The RSVM virtual machine instances78cand78dmay be configured to perform all, or any portion, of the techniques for improved rendition switching and/or any other of the disclosed techniques in accordance with the present disclosure and described in detail above. As should be appreciated, while the particular example illustrated inFIG. 7includes one RSVM virtual machine in each server, this is merely an example. A server may include more than one RSVM virtual machine or may not include any RSVM virtual machines.

Referring toFIG. 7, network203may, for example, be a publicly accessible network of linked networks and possibly operated by various distinct parties, such as the Internet. In other embodiments, network203may be a private network, such as a corporate or university network that is wholly or partially inaccessible to non-privileged users. In still other embodiments, network203may include one or more private networks with access to and/or from the Internet.

Network203may provide access to computers72. User computers72may be computers utilized by users70or other customers of data center75. For instance, user computer72aor72bmay be a server, a desktop or laptop personal computer, a tablet computer, a wireless telephone, a personal digital assistant (PDA), an e-book reader, a game console, a set-top box or any other computing device capable of accessing data center75. User computer72aor72bmay connect directly to the Internet (e.g., via a cable modem or a Digital Subscriber Line (DSL)). Although only two user computers72aand72bare depicted, it should be appreciated that there may be multiple user computers.

User computers72may also be utilized to configure aspects of the computing resources provided by data center75. In this regard, data center75might provide a gateway or web interface through which aspects of its operation may be configured through the use of a web browser application program executing on user computer72. Alternately, a stand-alone application program executing on user computer72might access an application programming interface (API) exposed by data center75for performing the configuration operations. Other mechanisms for configuring the operation of various web services available at data center75might also be utilized.

Servers76shown inFIG. 7may be servers configured appropriately for providing the computing resources described above and may provide computing resources for executing one or more web services and/or applications. In one embodiment, the computing resources may be virtual machine instances78. In the example of virtual machine instances, each of the servers76may be configured to execute an instance manager73aor73b(which may be referred herein singularly as instance manager73or in the plural as instance managers73) capable of executing the virtual machine instances78. The instance managers73may be a virtual machine monitor (VMM) or another type of program configured to enable the execution of virtual machine instances78on server76, for example. As discussed above, each of the virtual machine instances78may be configured to execute all or a portion of an application.

In the example data center75shown inFIG. 7, a router71may be utilized to interconnect the servers76aand76b. Router71may also be connected to gateway74, which is connected to network203. Router71may be connected to one or more load balancers, and alone or in combination may manage communications within networks in data center75, for example, by forwarding packets or other data communications as appropriate based on characteristics of such communications (e.g., header information including source and/or destination addresses, protocol identifiers, size, processing requirements, etc.) and/or the characteristics of the private network (e.g., routes based on network topology, etc.). It will be appreciated that, for the sake of simplicity, various aspects of the computing systems and other devices of this example are illustrated without showing certain conventional details. Additional computing systems and other devices may be interconnected in other embodiments and may be interconnected in different ways.

In the example data center75shown inFIG. 7, a server manager77is also employed to at least in part direct various communications to, from and/or between servers76aand76b. WhileFIG. 7depicts router71positioned between gateway74and server manager77, this is merely an exemplary configuration. In some cases, for example, server manager77may be positioned between gateway74and router71. Server manager77may, in some cases, examine portions of incoming communications from user computers72to determine one or more appropriate servers76to receive and/or process the incoming communications. Server manager77may determine appropriate servers to receive and/or process the incoming communications based on factors such as an identity, location or other attributes associated with user computers72, a nature of a task with which the communications are associated, a priority of a task with which the communications are associated, a duration of a task with which the communications are associated, a size and/or estimated resource usage of a task with which the communications are associated and many other factors. Server manager77may, for example, collect or otherwise have access to state information and other information associated with various tasks in order to, for example, assist in managing communications and other operations associated with such tasks.

It should also be appreciated that data center75described inFIG. 7is merely illustrative and that other implementations might be utilized. It should also be appreciated that a server, gateway or other computing device may comprise any combination of hardware or software that can interact and perform the described types of functionality, including without limitation: desktop or other computers, database servers, network storage devices and other network devices, PDAs, tablets, cellphones, wireless phones, pagers, electronic organizers, Internet appliances, television-based systems (e.g., using set top boxes and/or personal/digital video recorders) and various other consumer products that include appropriate communication capabilities.

As set forth above, content may be provided by a content provider to one or more clients. The term content, as used herein, refers to any presentable information, and the term content item, as used herein, refers to any collection of any such presentable information. A content provider may, for example, provide one or more content providing services for providing content to clients. The content providing services may reside on one or more servers. The content providing services may be scalable to meet the demands of one or more customers and may increase or decrease in capability based on the number and type of incoming client requests. Portions of content providing services may also be migrated to be placed in positions of lower latency with requesting clients. For example, the content provider may determine an “edge” of a system or network associated with content providing services that is physically and/or logically closest to a particular client. The content provider may then, for example, “spin-up,” migrate resources or otherwise employ components associated with the determined edge for interacting with the particular client. Such an edge determination process may, in some cases, provide an efficient technique for identifying and employing components that are well suited to interact with a particular client, and may, in some embodiments, reduce the latency for communications between a content provider and one or more clients.

In addition, certain methods or process blocks may be omitted in some implementations. The methods and processes described herein are also not limited to any particular sequence, and the blocks or states relating thereto can be performed in other sequences that are appropriate. For example, described blocks or states may be performed in an order other than that specifically disclosed, or multiple blocks or states may be combined in a single block or state. The example blocks or states may be performed in serial, in parallel or in some other manner. Blocks or states may be added to or removed from the disclosed example embodiments.

Although the flowcharts and methods described herein may describe a specific order of execution, it is understood that the order of execution may differ from that which is described. For example, the order of execution of two or more blocks or steps may be scrambled relative to the order described. Also, two or more blocks or steps may be executed concurrently or with partial concurrence. Further, in some embodiments, one or more of the blocks or steps may be skipped or omitted. It is understood that all such variations are within the scope of the present disclosure.

It should be emphasized that the above-described embodiments of the present disclosure are merely possible examples of implementations set forth for a clear understanding of the principles of the disclosure. Many variations and modifications may be made to the above-described embodiment(s) without departing substantially from the spirit and principles of the disclosure. The various features and processes described above may be used independently of one another, or may be combined in various ways. All possible combinations and subcombinations are intended to fall within the scope of this disclosure.

In addition, conditional language, such as, among others, “can,” “could,” “might,” or “may,” unless specifically stated otherwise, or otherwise understood within the context as used, is generally intended to convey that certain embodiments include, while other embodiments do not include, certain features, elements and/or steps.

Although this disclosure has been described in terms of certain example embodiments and applications, other embodiments and applications that are apparent to those of ordinary skill in the art, including embodiments and applications that do not provide all of the benefits described herein, are also within the scope of this disclosure. The scope of the inventions is defined only by the claims, which are intended to be construed without reference to any definitions that may be explicitly or implicitly included in any incorporated-by-reference materials.