Data placement for fault tolerance

[0001] The present invention relates to the field of data placement in distributed computing systems. More particularly, the present invention relates to data placement for fault tolerance in distributed computing systems.

[0002] Many computing systems take the form of a distributed system in which processing capability and data storage capacity are distributed in a network. Typically, nodes of the network each include a processor and data storage. The data placed at any given node may be accessed locally, for use by the processor at the node, or remotely, for use by the processor of another node.

[0003] In such a distributed environment, consideration must be given to the placement and replication of data in the system. At one extreme, a single copy of a data object, such as a file, may be placed in the network. However, if many nodes need to access the data object, such an arrangement may result in unacceptably high levels of traffic directed to the node having the single copy of the data and may also result in unacceptably low reliability should a failure occur at that node. At another extreme, the data object may be replicated at every node. This arrangement, however, may result in an unacceptable level of traffic associated with updating the data at every node in order to maintain consistency whenever changes to the data occur and may also result in unacceptable cost of storage capacity needed to store all the data copies in the system.

[0004] For a distributed system, the ideal placement of data will depend upon a variety of factors, such as the number and sizes of data objects, the frequency in which objects are accessed or changed, the number and distance between nodes that require access to the same data, desired performance levels and desired tradeoffs between cost, performance and reliability. Due to the complexity of the problem, algorithms for data placement have been largely based on heuristics.

[0005] In addition, prior algorithms for determining a placement of data in a distributed system have typically ignored the issue of fault-tolerance. Accordingly, such systems are unlikely to meet performance requirements in the event of failures that inevitably occur in a distributed system. While some prior algorithms have taken fault-tolerance into account, these solutions also have disadvantages in that they tend to be computationally intractable for systems having more than a few nodes. Another disadvantage is that such solutions have been applicable only to a specific type of distributed system and load and, thus, lack general applicability.

[0006] Therefore, what is needed is a technique for data placement in a distributed system that has more general applicability than prior techniques and that takes into account fault-tolerance. It is to these ends that the present invention is directed.

[0007] The invention is a technique for data placement in a distributed system that takes into account fault tolerance. Data placement is performed in which data objects, and possibly replicas thereof, are assigned to nodes within the distributed system. The resulting placement is then analyzed to determine whether the system provides desired performance under various different fault scenarios. If not, the distributed system is altered such as by altering its capacity or its capacity allocations. Performing data placement, analyzing for fault-tolerance and altering capacity or capacity allocations are performed repetitively, thereby increasing the system's ability to provide the desired performance under the fault scenarios. Preferably, a system and placement are eventually determined that provide the desired performance under the given fault scenarios.

[0012] The present invention is a technique that takes fault tolerance into account when placing data objects and replicas of data objects in a distributed computing system. In a distributed computing system, processing capability and data storage capacity are distributed in a network. Typically, nodes of the network each include a processor and data storage. Some nodes store data objects and, thus, function as storage servers. Other nodes access the data objects stored on the servers and, thus, function as clients of the servers. A node may function as both a storage server and as a client at the same time.

[0013] In accordance with the present invention, data objects are placed in a distributed system to satisfy desired performance levels, even under fault conditions. The distributed system may be modeled (e.g., analytical model or a simulation model), in which case, representations of data objects may be placed in a model of a distributed system. The data objects, and possibly replicas thereof, are placed by assigning each to a node within the distributed system (or model). A conventional data placement algorithm, including one that does not take fault-tolerance into account, may be used to initially place the data objects. The placement is then tested under various failure scenarios to determine whether the desired performance is achieved even under these failures. For example, where the conventional placement algorithm does not take into account the possibility of a server failure, such a failure may result in requests being redirected to another server. This redirection, however, may violate the desired performance level for the system (e.g., responses to the redirected requests may take too long due to server congestion or longer network distances). Where such a problem is encountered, the invention iteratively alters the infrastructure of the distributed system and places the objects based on the altered infrastructure in an attempt to provide the desired performance level despite the failure. As a result, the data objects are placed in such a way as to provide desired performance levels despite failures.

[0014] FIG. 1 illustrates an exemplary distributed computing system 100 for which data placement may be performed in accordance with the present invention. As shown in FIG. 1, each of several nodes 102 includes a processor 104 and data storage 106. The nodes 102 are interconnected by communication links 108 of a network 110. The distributed system 100 is exemplary in that the invention is applicable to any distributed system that allows flexibility in placement of data objects. Such systems include, but are not limited to, Storage Area Networks (SANs), Network Attached Storage (NAS) systems, storage bricks, global file systems, service allocation in and between data centers, content delivery networks, and so forth. Accordingly, the network 110 and the number and type of nodes 102 can differ from that illustrated in FIG. 1. For example, while the nodes 102 may be identical to each other, this will generally not be the case. Rather, a variety of different nodes, having different types of processors, different types and capacities of data storage, different types of network interfaces, etc., may be included in the same distributed system 100. Further, some nodes may function as storage servers only and may have limited processing capability to perform other tasks.

[0015] FIG. 2 illustrates diagrammatically a method 200 for performing data placement in accordance with the present invention. The method receives as input an initial infrastructure specification of the distributed system 100. The infrastructure specification specifies the number of nodes 102 in the system 100 and interconnections among the nodes. The infrastructure specification may include information for each node, for example, its data storage capacity, its processor capacity and its input/output (I/O) throughput and latency for data storage operations. In addition, the infrastructure specification may include information for each communication link 108 between nodes, such as communication bandwidth and latency.

[0016] In addition, the method 200 receives as input a workload specification for the distributed system 100. For example, the workload specification may include a trace (i.e. a listing) of activity records for the system 100 in which each activity record specifies an operation within the system 100 that generates storage or communication activity. For example, an activity record for a storage operation may identify an initiator of the operation (e.g., a particular application), a particular data object to which the operation is directed, the type of operation (e.g., read, write, refresh, etc.), and the location(s) within the data object of the data affected by the operation. As another example, an activity record for a communication operation may identify an initiator of the communication (e.g., a particular application), a recipient of the communication (e.g., another application or a data object) and the quantity of data communicated. The workload specification also specifies initial conditions for the distributed system 100, such as a number and size of data objects to be stored within the distributed system 100 and applications to be served by the system 100. The workload specification may be collected from the operation of a distributed system or may be artificially generated using conventional techniques for generation of a synthetic workload.

[0017] Performance objectives may also be received as input to the method 200. The performance objectives specify desired levels of performance to be achieved by the distributed system 100 as it operates under the workload. For example, a particular performance objective may specify a maximum latency allowable for certain operations within the system 100. For example, a maximum latency may be specified for data retrieval requests where the retrieval time is the time required to deliver data to a requesting application measured from the time the request is issued by the application. Another performance objective may specify that less than a predetermined percentage of requests (e.g., 10%) exceed a specified latency. Yet another performance objective may specify a throughput objective, such as a maximum queue length for sequential operations that are waiting to occur after prior operations in the sequence are completed.

[0018] Further, one or more of the performance objectives, and possibly all of the performance objectives, are specified to be met despite the occurrence of certain faults within the system 100, such as inoperable communication links or storage devices. Thus, a failure specification may be provided as input to the method 200. The failure specification indicates the failures that the system 100 is intended to be able to withstand while still maintaining the performance objectives. For example, the failure specification may specify that the system 100 is to withstand a complete failure of any single network element, such as a node or communication link. As another example, the failure specification may specify that the system 100 is to withstand the simultaneous failure of any two (or more) nodes or communication links (or combination thereof).

[0019] An initial data placement is performed in step 202, based on input to the method 200, such as the infrastructure specification and the number of type of data objects to be placed in the distributed system. The initial data placement includes the data objects, and possibly replicas of the data objects, assigned to various locations within the system 100. For example, a data object may be assigned to a node or to a storage device of a node, depending on the specificity of the infrastructure specification and placement algorithm. A conventional data placement algorithm may be used in step 202 that does not take into account fault tolerance objectives.

[0020] One exemplary algorithm that may be used in step 202 may be referred to as a “greedy ranking algorithm.” Using this algorithm, all possible combinations of placing one copy of a data object at one node in the system are ranked. The ranking is in accordance with criterion that is expected to affect the performance of the system 100, such as a number of local read operations required for a particular placement. The data object with the highest rank is then placed in the system 100. A new ranking is then computed for the remaining data objects and the one having the highest rank is placed. This process is repeated until some predefined objective has been met. This could be, for example, that all of the objects have been placed or the desired performance requirements are satisfied. Another exemplary algorithm that may be used in step 202 may be referred to as a “swap algorithm.” Using this algorithm, an initial placement is generated, for example, by a random placement. Then, positions of pairs or groups of data objects are exchanged (i.e. swapped) until a placement is achieved that satisfies desired performance requirements. While the greedy ranking algorithm or the swap algorithm may be used in step 202, it will be apparent that another placement algorithm may be used.

[0021] In step 204, one or more failure scenarios are applied to the system 100 and a determination is made as to whether or not the specified performance objectives are met under each failure scenario. If the specified performance objectives are not met under an applied failure scenario, then program flow moves to a step 206 in which the initial infrastructure specification is altered. This may be accomplished by altering capacity and/or capacity allocation of a resource in the infrastructure specification.

[0022] Capacity may be altered, for example, by increasing or decreasing the number of nodes in the system, increasing or decreasing the storage capacity at one or more nodes (e.g., by altering the number or size of storage devices, such as hard disks), by increasing or decreasing the number of communication links among nodes, or by increasing or decreasing the bandwidth capacity of one or more of the communication links.

[0023] Capacity, such as storage capacity of a node or bandwidth capacity of a communication link may be allocated to certain uses. For example, storage capacity may be allocated to a particular data object or application. Similarly, bandwidth capacity of communication links may be allocated to particular application programs. These allocations of storage capacity and communication bandwidth tend to inhibit applications from interfering with each other. For example, the allocation for storage space on a node may be in excess of the actual size of the objects stored there so as to allow an application's storage demands to grow over time. These allocations may be provided, for example, as input to the method 200 or may be developed during the data placement step 202. Thus, altering capacity allocated to a use tends to alter the capacity available for other uses. Accordingly, rather than, or in addition to, altering capacity, capacity allocation may be altered in step 206.

[0024] Capacity allocations may be altered, for example, by altering a maximum allowed bandwidth utilization of a communication link between a pair of nodes of the distributed system 100 or by altering a maximum allowed storage capacity utilization of a node of the distributed system 100.

[0025] Increasing capacity will tend to increase the cost of the distributed system because additional hardware (e.g., storage devices) will generally be required to provide the additional capacity. However, increasing capacity allocation, which generally does not require additional hardware, is generally expected to have a lower associated cost. Accordingly, alterations to capacity allocation may be attempted before alterations to capacity, so as to minimize cost.

[0026] In general, it is expected that increases in capacity or capacity allocation are more likely to improve the fault tolerance of the distributed system than decreases. However, in certain circumstances it may be desired to also decrease capacity or capacity allocation, particularly where a decrease is accompanied by an offsetting increase in capacity or capacity allocation.

[0027] In one aspect, where the system 100 fails to meet a performance objective in step 204, a node or a communication link is added to the system 100 in step 206 before returning to the data placement step 202. Determining whether to add a node or a communication link may be based on which is in shortest supply. Thus, for example, if a performance objective that is not met is for communication latency between nodes, then an additional communication link may be added to the system 100. However, if a performance objective that is not met is for throughput of storage operations, then an additional node may be added to the system 100.

[0028] After altering the infrastructure in step 206, placement is performed in step 202 based on the newly-changed infrastructure. The same placement algorithm may be performed in each pass through the step 202. Thus, in step 202 all of the data objects may be reassigned to the newly-changed infrastructure. Alternately, for certain passes through the step 202, a simplified placement algorithm may be performed. For example, a portion the data objects may re-assigned to the newly-changed infrastructure during a pass through step 202, rather than re-assigning all of the objects during each pass.

[0029] FIG. 3 illustrates a flow diagram 204 for performing failure checking in accordance with an aspect of the present invention. FIG. 3 illustrates step 204 of FIG. 2 in more detail. From step 202 (FIG. 2), program flow begins in step 302 in which a first failure scenario to be tested is generated. Thus, a failure scenario, such as a link that is down or an inoperable node, is applied to the distributed system 100. This may be accomplished by disabling a portion of the system 100 so as to simulate a failure of the disabled portion.

[0030] In step 304, the system 100 is operated under the failure conditions applied in step 304 to determine whether the system 100 meets the desired performance objectives under the failure conditions. This may include operating a real distributed system or a computer-implemented model thereof. Thus, in step 304 the workload specification may be applied to the system 100 while portions of the system 100 are disabled. During operation, system performance is monitored, such as by measuring latencies of various operations or queue lengths for operations that are scheduled to occur.

[0031] In step 306, a determination is made as to whether the performance objectives are satisfied under the current failure scenario. When the performance objectives are satisfied, program flow moves to a step 308 in which a determination is made as to whether all desired failure scenarios have been tested. For example, a set of failure scenarios may encompass all possible single failures of a node or communication link. Thus, each failure scenario may specify one link or node that is to be disabled for testing the system 100. In other examples, it may be desired to determine whether the system 100 is able to withstand multiple simultaneous failures. In this case, a set of failure scenarios may specify various combinations of two or more links or nodes that are to be simultaneously disabled or all possible combinations of such failures.

[0032] When all the failure scenarios have not yet been tested, program flow moves to a step 310. In step 310, a next failure scenario is generated. In this step, a portion of the distributed system 100 may be disabled that is different from the portion disabled in step 302.

[0033] In this way, each desired failure scenario may be tested sequentially. The sequence may be random, or predetermined, e.g., by the failure specification. Alternately, the sequence may be changed depending upon prior results. For example, previously failed scenarios may be tested prior to untested scenarios or previously passed scenarios. As another example, failure scenarios may be ordered according which is most likely to result in failure to meet a performance objective. Thus, scenarios in which one or more heavily-loaded elements, such as nodes and/or communication links, are disabled may be tested before scenarios in which lightly-loaded elements are disabled. As another example, once a failure scenario is passed by the design, it may be omitted from all subsequent testing.

[0034] From step 310, program flow returns to step 304 in which the distributed system (real or modeled) is operated again under the workload, however, for this pass through step 304 new failure conditions are applied to the system 100.

[0035] This process of generating failure scenarios and testing the system 100 to determine whether the performance objectives are met may be repeated until it is determined that a performance objective is not met for a failure scenario or until it is determined that all of the scenarios have been successfully tested.

[0036] When it is determined that a performance objective is not satisfied (in step 306), program flow may return to step 206 of FIG. 2 in which the infrastructure specification is altered in an attempt to provide an ability to withstand the particular failure which caused the performance objectives to not be met. Once the infrastructure specification has been altered in step 206 (FIG. 2) and a new placement made in step 202 (FIG. 2), program flow may return to step 302.

[0037] In a next pass through the method of FIG. 3, all of the same failure scenarios may be repeated so as to ensure that the altered system 100 still satisfies the performance objectives under all of the failure scenarios. For example, where a capacity or a capacity allocation is reduced in step 206, this may result in a failure scenario under which the performance objectives were previously determined to be met to no longer be met. Thus, it may be desired to re-test all failure scenarios.

[0038] Alternately, in the next pass through the method of FIG. 3, only scenarios that previously caused the performance objectives to not be met (i.e. the failed scenarios) may be performed so as to avoid repeating tests that have already been passed. For example, where capacity or capacity allocations are increased only, it is expected to be unlikely that a failure scenario that resulted in performance objectives being met (i.e. a passed scenario) will now result in a performance objective not being met.

[0039] When it is determined in step 308, that all of the performance objectives are satisfied under each of the failure scenarios, this indicates that a data placement solution has been found. In this case, program flow may terminate with this solution. The method 100 may be repeated (using the same or different placement algorithms), so that a design solution can be selected from among multiple alternative solutions. For example, a lowest-cost solution may be selected from the alternatives.

[0040] Thus, a technique has been described in which fault tolerance is taken into account for data placement. The invention achieves significant advantages in that can be performed efficiently in that it can be performed programmatically by use of a computer system. In addition, the invention is applicable to wide variety of distributed systems.

[0041] FIG. 4 illustrates a block schematic diagram of general-purpose computer system by which the present invention, including portions of the methods of FIGS. 2 and 3, may be implemented. The computer system 400 may include a general-purpose processor 402, a storage media 404, such as persistent memory (e.g., a hard disk for program memory) and transitory memory (e.g., RAM), a communication bus 406, and input/output devices 408, such as a keyboard, monitor, mouse, printer and network interface. The computer system 400 is conventional. As such, it will be apparent that the system 400 may include more or fewer elements than shown in FIG. 4 and that other elements may be substituted for those illustrated in FIG. 4.

[0042] Thus, the inputs to the method 200, such as the infrastructure specification, workload specification and performance guarantees may be input to the computer system 400 in the form of computer-readable data files that may be stored in computer-readable media 404. In addition, the computer system 400 may implement the methods of FIGS. 2 and 3 by the CPU 402 executing a programmed sequence of instructions stored in the computer-readable media 404.

[0043] As mentioned, the method 200 may be performed on a real distributed system, as is illustrated in FIG. 1. Alternately, the method 200 may be performed on a model of the distributed system, such as an analytical model or a simulation model. Where the distributed system 100 is modeled, the model may be implemented, for example, by the computer system 400. A real system may then be constructed based on results using the model. The methods of FIGS. 2 and 3 may then be applied to the real system so constructed. In this way, the results based on the model can be verified and refined, as necessary, using the methods of FIGS. 2 and 3.

[0044] While the foregoing has been with reference to particular embodiments of the invention, it will be appreciated by those skilled in the art that changes in these embodiments may be made without departing from the principles and spirit of the invention, the scope of which is defined by the appended claims.