text:
Liana Evans | August 18, 2010 | Comments
Companies tend to forget that they need to actually go out and engage in their social media communities.
As a social media consultant for my clients, I occasionally get sent links to some bizarre tools, services, and "odd stuff" to look at, examine, and check out for validity. I was just sent a link to a service that automatically generated "social media" messages for people who signed up for their service and I had to pause and think about how this could actually do more harm than good to any serious social media marketing campaign.
Everyone wants to jump on the social media marketing bandwagon these days. Why wouldn't you? With all the success of Old Spice, Ford, Whole Foods, Dell, and many more, it makes sense to go that route. If you look at it from a very high level view, you might think these feats of accomplishment were easy things by these companies. But it's far from easy.
As social media marketing and the sites and communities people are sharing in become more mature, it takes a lot more to find success. It's not as simple as setting up a profile in Facebook and hoping that your fans will come. You can't just toss a video onto YouTube and hope for viral success, no matter how good looking the guy in the towel is you have starring in the video. There has to be some serious research, strategy, measurement, and engaging going on.
Engagement is probably the key to it all. After all, social media does have the word "social" in it. Just pushing your messages out there, loading up your commercials, or putting your marketing slicks on Scribd is not likely to get you far, no matter how much someone tells you that you "need" to have an account. The part of the "set and forget it" equation with social media that agencies and companies tend to forget is that you need to actually go out and engage in these communities.
There are many elements to why engagement is important. From search engines finding relevancy to social media users actually knowing about your presence, engagement has to be a central piece to any social media strategy. Just as important as measuring what you're doing and understanding who you're targeting, making sure you're sharing and engaging with the communities you're looking to market in is essential.
With that all said, I don't suggest jumping in immediately and acting like a know-it-all in your chosen forum. There's a lot of advantage to first researching your community before you start trying to be a social butterfly. By first observing and listening to the community, you can identify your influencers, detractors, and the "normal Joes" who are active in the community, and choose how best to engage with them before you even type one word.
There are some automatic tools out there, like the ones I mentioned earlier, that seem to make implementation of your social media strategy a lot more efficient. The trouble with these tools is that the automatic generation isn't actually from you or your company, and eventually the audience will find out that it's not you engaging. If you set up the expectation that it's members of your company's staff that are engaging and then switch to an automatic method, people will notice and all your hard work will be for nothing.
That's not to say that tools like Dlvr.it don't have their place. These tools are extremely helpful to any social media strategy, especially one where time is a valuable resource and it's better spent with your employees actually engaging rather than posting your latest blog posts on your Facebook wall or Twitter stream. It's a point of expectations that you set with your audience.
Audiences are not keen on being "tricked" and they will shout from the mountain tops if they suspect that anything is out of the ordinary or has changed for what they perceive as the worse. That's why in your social media strategy it's important to set the right expectations when you start engaging with the audience.
Social media can be very resource-intensive and that's why automated social media tools can be appealing. Some companies make the mistake of starting off "gangbusters" in social media and finding out they cannot maintain that level of activity in the community, and turn to tools like these to try and solve the issue.
Answering questions, asking opinions, and sharing thoughts and ideas are all human engagement opportunities with your target audience that you can't get from automation or by setting up a profile in a social media community. Being "social" in social media can lead to a lot of success if planned out properly…just resist the urge to try and totally automate your strategy - it can lead to more failures than you ever anticipated.
Want to learn more?
Attend ClickZ Live New York March 30 - April 1. With over 15 years' experience delivering industry-leading events, ClickZ Live brings together over 60 expert speakers to offer an action-packed, educationally-focused agenda covering all aspects of digital marketing. Register today!
Liana "Li" Evans is the author of the award winning social media marketing book, "Social Media Marketing: Engaging Strategies for Facebook, Twitter & Other Social Media" and she is the president and CEO of Da Li Social, as well as an adjunct professor for Rutgers University's Mini MBA Program. Liana has also been featured in the books "Online Marketing Heroes" and "Video Marketing An Hour a Day." As an established online marketing industry veteran with over 15 years of experience she's focused her unique skillset to specialize in integrated marketing and how companies can successfully strategize integrating all online marketing channels as well as offline traditional media. Her deep technical combined with a public relations background enables her to partner with clients for establishing successful online marketing campaigns that combine cross-channel tactics cohesively.
Li was the search engine optimization (SEO) and social media marketing architect for such companies as QVC and Comcast (Fancast) and has consulted with several other different sized companies such as AOL MovieFone. Her wealth of knowledge in dealing with large e-commerce and content sites allows her a wider perspective into what it takes to launch successful marketing campaigns in the online space.
March 24-25, 2015
March 26, 2015
Mar 30-Apr 1, 2015
May 26-28, 2015
June 23-25, 2015
Singapore, 5-6 March
Bangkok, 17-18 March
Hong Kong, April 2015
A Buyer's Guide to Affiliate Management Software
Manage your performance marketing with the right solution. Choose a platform that will mutually empower advertisers and media partners!
Google My Business Listings Demystified
To help brands control how they appear online, Google has developed a new offering: Google My Business Locations. This whitepaper helps marketers understand how to use this powerful new tool.
March 3, 2015
1:00pm ET/10:00am PT


text:
US 20070277056 A1 Abstract A distributed system for creating a checkpoint for a plurality of processes running on the distributed system. The distributed system includes a plurality of compute nodes with an operating system executing on each compute node. A checkpoint library resides at the user level on each of the compute nodes, and the checkpoint library is transparent to the operating system residing on the same compute node and to the other compute nodes. Each checkpoint library uses a windowed messaging logging protocol for checkpointing of the distributed system. Processes participating in a distributed computation on the distributed system may be migrated from one compute node to another compute node in the distributed system by re-mapping of hardware addresses using the checkpoint library.
Claims(64) 1-46. (canceled)
47. A compute node capable of operating as part of a distributed system, comprising:
memory; and a processor configured to access the memory to perform a process in a distributed computation running on the distributed system, the processor being further configured to record a first set of memory locations modified by the processor during a first checkpoint interval, and create a checkpoint from the contents of the first set of memory locations, while recording a second set of memory locations modified by the processor during a second checkpoint interval. 48. The compute node of
49. The compute node of
50. The compute node of
51. The compute node of
52. The compute node of
53. The compute node of
54. The compute node of
55. The compute node of
56. The compute node of
57. The compute node of
58. A compute node capable of operating as part of a distributed system, comprising:
memory; and a processor configured to perform a process in a distributed computation running on the distributed system, the processor being further configured to store in the memory a copy of each message output from the compute node until an acknowledgement is received, the processor being further configured to create a checkpoint, and if a subsequent failure occurs, roll the compute node back to the checkpoint and output each message copied in the memory that does not receive an acknowledgement after the compute node is rolled back to the checkpoint. 59. The compute node of
60. A compute node capable of operating as part of a distributed system, comprising:
a processor configured to perform a process in a distributed computation running on the distributed system, the processor being further configured to receive messages, and output an acknowledgement for each message received, the processor being further configured to create a checkpoint, and if a subsequent failure occurs, roll the compute node back to the checkpoint, recognize and discard duplicate messages received by the compute node after the compute node is rolled back to the checkpoint, and for each duplicate message, output an acknowledgement. 61. The compute node of
62. A compute node capable of operating as part of a distributed system, comprising:
a processor configured to perform a process in a distributed computation running on the distributed system, the processor being further configured to create a checkpoint for the process, and in response to a preemptive scheduling request, store the checkpoint to non-volatile memory and halt the process. 63. The compute node of
64. The compute node of
65. The compute node of
66. Computer readable media embodying a program of instructions executable by a processor to perform a method of creating a checkpoint for a process in a distributed computation running on a distributed system, the method comprising:
recording a first set of memory locations modified by the process during a first checkpoint interval; and creating a checkpoint from the contents of the first set of memory locations, while recording a second set of memory locations modified by the process during a second checkpoint interval. 67. The computer readable media of
68. The computer readable media of
69. The computer readable media of
70. The computer readable media of
71. The computer readable media of
72. The computer readable media of
73. The computer readable media of
74. The computer readable media of
75. The computer readable media of
76. The computer readable media of
77. Computer readable media embodying a program of instructions executable by a processor to perform a method of creating a checkpoint for a process in a distributed computation running on a distributed system, the process being performed by a compute node, the method comprising:
storing in a memory a copy of each message output from the compute node until an acknowledgement is received; creating a checkpoint for the process; rolling compute node back to the checkpoint in response to a failure; and outputting each message copied in the memory that does not receive an acknowledgement after the compute node is rolled back to the checkpoint. 78. The computer readable media of
79. Computer readable media embodying a program of instructions executable by a processor to perform a method of creating a checkpoint for a process in a distributed computation running on a distributed system, the process being performed by a compute node, the method comprising:
outputting an acknowledgement for each message received by the compute node; creating a checkpoint for the process; rolling the compute node back to the checkpoint in response to a failure; and recognizing and discarding duplicate messages received by the compute node after the compute node is rolled back to the checkpoint, and for each duplicate message, outputting an acknowledgement. 80. The computer readable media of
81. Computer readable media embodying a program of instructions comprising a checkpoint library executable by a processor having access to an operating system and a distributed application to perform a process in a distributed computation running on a distributed system, the checkpoint library comprising:
instructions to create a checkpoint for the process, the creation of the checkpoint being transparent to the operating system and the distributed application. 82. The computer readable media of
83. The computer readable media of
84. Computer readable media embodying a program of instructions executable by a processor to perform a method of creating a checkpoint for a process in a distributed computation running on a distributed system, the method comprising:
creating a checkpoint for the process; and storing the checkpoint to non-volatile memory and halting the process in response to a preemptive scheduling request. 85. The computer readable media of
86. The computer readable media of
87. The computer readable media of
88. A method of creating a checkpoint for a process in a distributed computation running on a distributed system, the method comprising:
recording a first set of memory locations modified by the process during a first checkpoint interval; and creating a checkpoint from the contents of the first set of memory locations, while recording a second set of memory locations modified by the process during a second checkpoint interval. 89. The method of
90. The method of
91. The method of
92. The method of
93. The method of
94. The method of
95. The method of
96. The method of
97. The method of
98. The method of
99. A method of creating a checkpoint for a process in a distributed computation running on a distributed system, the process being performed by a compute node, the method comprising:
storing in a memory a copy of each message output from the compute node until an acknowledgement is received; creating a checkpoint for the process; rolling compute node back to the checkpoint in response to a failure; and outputting each message copied in the memory that does not receive an acknowledgement after the compute node is rolled back to the checkpoint. 100. The method of
101. A method of creating a checkpoint for a process in a distributed computation running on a distributed system, the process being performed by a compute node, the method comprising:
outputting an acknowledgement for each message received by the compute node; creating a checkpoint for the process; rolling the compute node back to the checkpoint in response to a failure; and recognizing and discarding duplicate messages received by the compute node after the compute node is rolled back to the checkpoint, and for each duplicate message, outputting an acknowledgement. 102. The method of
103. A method of creating a checkpoint for a process in a distributed computation running on a distributed system, the method comprising:
creating a checkpoint for the process; and storing the checkpoint to non-volatile memory and halting the process in response to a preemptive scheduling request. 104. The method of
105. The method of
106. The method of
107. A method of migrating a process in a distributed computation running on a distributed system from a first compute node to a second compute node in the distributed system, each of the first and second compute nodes having an operating system, the method comprising:
creating a checkpoint for the process in the first compute node; and migrating the process to the second compute node by providing the second compute node with the checkpoint without migrating the operating system from the first compute node to the second compute node. 108. The method of
109. The method of
Description This application is a non-provisional application claiming benefit under 35 U.S.C. sec. 119(e) of U.S. Provisional Application Ser. No. 60/520,254, filed Nov. 17, 2003 (titled TRANSPARENT CHECKPOINTING AND MIGRATION OF PARALLEL CODES OVER GRID INFRASTRUCTURES by Srinidhi Varadarajan), which is hereby incorporated in full by reference herein. The present disclosure generally relates to distributed computing systems, and more particularly, to improving the checkpointing of distributed computations executed on such systems. Cluster supercomputing is the practice of connecting individual computing nodes to create a distributed system that provides a computing resource capable of solving complex problems. These nodes may be individual desktop computers, servers, processors or similar machines capable of hosting an individual instance of computation. These nodes are constructed out of hardware components including, but not limited to, processors, volatile memory (RAM), magnetic storage drives, mainboards, network interface cards, etc. There has been a thrust recently in the HPC (High Performance Computing) community towards utilizing distributed systems as opposed to the more traditional custom supercomputers. This movement has largely been motivated by the relatively recent availability of high speed network interconnects (e.g., Myrinet, Quadrics, and Infiniband) that allow distributed systems to reach similar levels of efficiency as those observed by traditional custom supercomputers at a fraction of the cost. Such systems still suffer from the major drawback of comparatively poor system reliability. Assuming for illustration that the average individual computing node C has a reliability of x, the probability that none of the hardware components that comprise C will fail on a given day. Often x is what appears to a very high probability, perhaps 99.9%. This represents excellent reliability for the normal consumer, who has no issue with having to perform maintenance on the single component approximately once a year. The quandary arises however, when one examines precisely how x behaves with regards to the probability of any single node Ci in the distributed system failing. The probability P of any node Ci failing in a group of n nodes is given by:
As n increases, the probability of a node failing on a given day increases linearly. Indeed, once n crests 1000, a not uncommon number of components for larger distributed systems, it is almost guaranteed that a minimum of one node will fail on a daily basis. This lack of reliability is further exacerbated by the fact that additional node failures are caused by imperfect system software. Any distributed computation that was utilizing the failed node would then have to be restarted. Many of the HPC applications which utilize large distributed systems take days or weeks, even months to complete, most likely several failed attempts would be required before a distributed computation manages to complete, if at all. As a result distributed systems unable to tolerate failures are unusable for truly large scale supercomputing. If there were a method to save the state of a distributed computation such that it could be restarted in that state after failures were resolved, then combining that method with a distributed system might result in a computing resource with the reliability of a traditional supercomputer, at a fraction of the cost. There have been numerous attempts to provide such a method, almost all of which fall into one of two abstract classifications: checkpoint-based protocols and log-based protocols. A comprehensive survey of both checkpoint-based and log-based protocols is available in E. N. (Mootaz) Elnozahy, Lorenzo Alvisi, Yi-Min Wang, and David B. Johnson. A Survey of Rollback-Recovery Protocols in Message-Passing Systems. ACM Comput. Surv., 34(3): 375408, 2002, which is incorporated herein by reference. The requirements to �checkpoint� or record the state of a single non-distributed computation is simple and well known. It involves merely recording the state (e.g., global data, stack, heap, mapped memory, and processor context) of the software process that realizes the computation, to some form of persistent storage. This data saved to persistent storage is known as a �checkpoint�. At a later time the checkpoint may be read from stable storage and loaded by a process, after which computation will transparently resume at the point of execution in the saved state. Periodic checkpointing of a long running computation allows for tolerance of failures. A computation can be restarted from its most recent checkpoint once the failure has been resolved. Utilizing this method the only part of the computation lost is that which took place in the interval between the most recent checkpoint and the failure. When one attempts to apply this same method to a distributed computation, however, the challenge becomes much more substantial. A distributed computation is one in which several instances of computation work in concert to solve a single problem. Each instance of computation or �process� is usually implemented as an individual OS process or a thread of execution inside an OS process. The cooperation between the separate processes takes the form of exchanged messages. These messages are exchanged either over an interconnection network or through the accessing and modification of shared memory. In order for a checkpoint of a distributed computation to be of use, it must represent a state that is globally consistent. A globally consistent state is one that could have been reached during the normal course of the execution of the computation. The difficulty in checkpointing a distributed computation lies in the fact that at any given time there are probably many messages �in-flight� between the different processes, implying that the communication channels possess state that must be captured. Consider a distributed computation comprised of two processes (Ps and Pr) at either end of a communication channel. Ps is checkpointed prior to sending a particular message m, while Pr is checkpointed after the receipt of m. The global state represented by the aggregate of the two checkpoints is not consistent because one process has received a message that the other process never sent. This phenomenon is referred to as an orphan message and demonstrates that in order to ensure that the checkpoint of a distributed computation is globally consistent there must be some level of coordination between the individual processes. Almost all conventional methods to checkpoint distributed computations are based on the method of Distributed Snapshots as described, for example, by K. Mani Chandy and Leslie Lamport. Distributed Snapshots: Determining Global States of Distributed Systems. ACM Trans. Comput. Syst., 3(1): 63-75, 1985, which is incorporated herein by reference. This method is a global state detection mechanism that achieves coordination through the use of �marker� messages. It relies on a fundamental assumption that the communication channels of the distributed system are reliable, FIFO (First-In First-Out) queues that guarantee all messages sent by one process to another are received in-order and without error. When a single process in such a distributed computation wishes to detect a global state (which can be recorded as a checkpoint) it sends a marker message out on all its communication channels and immediately records its local state. Each process on the other end of a communication channel receives the marker message and records its local state. The process then forwards the marker message on each channel with the exception of the channel on which the marker was received. These marker messages propagate throughout the distributed system and coordinate the checkpointing of individual processes such that the aggregate of all the individual checkpoints equates to a globally consistent state. In order to understand how this coordination is accomplished, consider again the case of a distributed system comprised of two processes and a single reliable FIFO communication channel connecting them. One of the two processes PS initiates a checkpoint by sending a marker message across the channel and recording its local state. Immediately upon receipt of the marker message, the receiving process Pr saves its local state. Pr guarantees it received all messages sent before Ps took a checkpoint. Additionally this guarantees guarantee that Pr's own checkpoint was taken before it received any messages sent by Ps after Ps checkpointed. The result is that when the two processes save their respective states no messages are sent but not yet received and no messages are received but not yet sent. In effect, the marker messages �flush�, or �drain�, the network of all messages so as to restrict the state of the distributed computation that must be recorded to that of the individual processes. This precludes any inconsistencies from arising upon restart. The LAM/MPI message passing library is one well-known communication middleware implementation that utilizes distributed snapshots to coordinate individual process checkpoints taken with Berkeley Linux Checkpoint Restart (BLCR), which is a single process kernel based checkpoint/restart system. The LAM/MPI message passing library is discussed further in Greg Burns, Raja Daoud, and James Vaigi. LAM: An Open Cluster Environment for MPI. In Proceedings of Supercomputing Symposium, pages 379-386, 1994, and also in Jeffrey M. Squyres and Andrew Lumsdaine. A Component Architecture for LAM/MPI. In Proceedings, 10th European PVM/MPI Users' Group Meeting, number 2840 in Lecture Notes in Computer Science, pages 379-387, Venice, Italy, September/October 2003 (Springer-Verlag), each of which is incorporated herein by reference. BLCR is described in more detail by J. Duell, P. Hargrove, and E. Roman. The Design and Implementation of Berkeley Lab's Linux Checkpoint/Restart, 2002, which is incorporated herein by reference. When the LAM library desires to record the state of a distributed computation, its drains the network of all messages utilizing the marker packets, shuts down all communication channels to remove any state from the OS, and utilizes BLCR to checkpoint the local state of each individual process. The foregoing is discussed further in Sriram. Sankaran, Jeffrey M. Squyres, Brian Barrett, Andrew Lumsdaine, Jason Duell, Paul Hargrove, and Eric Roman. The LAM/MPI checkpoint/restart framework: System-Initiated Checkpointing. In Proceedings, LACSI Symposium, Sante Fe, N.M., USA, October 2003, which is incorporated herein by reference. The LAM library then reopens all communications channels and continues computation. Accordingly, there are several drawbacks and shortcomings shared by current implementations of distributed checkpoint/restart based on the distributed snapshots method. Most current methods suffer from one or more of the following disadvantages:
For a more complete understanding of the present disclosure, reference is now made to the following figures, wherein like reference numbers refer to similar items throughout the figures: The exemplification set out herein illustrates particular embodiments, and such exemplification is not intended to be construed as limiting in any manner. The following description and the drawings illustrate specific embodiments sufficiently to enable those skilled in the art to practice the systems and methods described herein. Other embodiments may incorporate structural, logical, process and other changes. Examples merely typify possible variations. Individual components and functions are optional unless explicitly required, and the sequence of operations may vary. Portions and features of some embodiments may be included in or substituted for those of others. The elements that implement the various embodiments of the present system and method are described below, in some cases at an architectural level. Many elements may be configured using well-known structures. The functionality and processes herein are described in such a manner to enable one of ordinary skill in the art to implement the functionality and processes described herein. The processing described below may be performed by a distributed processing platform. In addition, such processing and functionality can be implemented in the form of special purpose hardware or in the form of software or firmware being run by a general-purpose or network processor. Data handled in such processing or created as a result of such processing can be stored in any type of memory as is conventional in the art. By way of example, such data may be stored in a temporary memory, such as in the RAM of a given computer system or subsystem. In addition, or in the alternative, such data may be stored in longer-term storage devices, for example, magnetic disks, rewritable optical disks, and so on. For purposes of the disclosure herein, a computer-readable media may comprise any form of data storage mechanism, including existing memory technologies as well as hardware or circuit representations of such structures and of such data. It should also be understood that the techniques of the present system and method might be implemented using a variety of technologies. For example, the methods described herein may be implemented in software running on a programmable microprocessor, or implemented in hardware utilizing either a combination of microprocessors or other specially designed application specific integrated circuits, programmable logic devices, or various combinations thereof. In particular, the methods described herein may be implemented by a series of computer-executable instructions residing on a storage medium such as a carrier wave, disk drive, or other computer-readable medium. A method as described herein for transparent, non-blocking, detection of the global state of a non-interactive distributed computation executing on some subset of the nodes in a distributed system that enables the migration of individual processes may be combined with a distributed system to create a computing resource that may provide reliability comparable to that of a traditional supercomputer built from custom hardware. The disclosure herein provides methods and apparatus that: may be used to guarantee the consistency of a communication channel's state during a distributed checkpoint, commits the recorded state without blocking the computation, and coordinates the local checkpointing of the individual processes in such a way that the distributed system may recover from failures without manual intervention or system stoppage requiring only minimal loss of computational efficiency. The disclosure provides these methods in such a manner to permit the migration of individual processes during the course of the distributed computation. Furthermore, an embodiment may take the form of a library that can be transparently interposed (e.g., through linking) with legacy applications, middleware, and OS's, without any modification of the legacy software. In an illustrative embodiment, a distributed system includes n computers, each of which possesses p processors, interconnected by a network. A non-interactive distributed computation is started on anywhere from 2 to n computers where each processor hosts an individual process. During the course of the distributed computation, the processes interact through the exchange of messages. This communication may be implemented by the user level application that is executing, but most likely is implemented by a middleware library (e.g., MPICH, MVAPICH, LAM/MPI) that is linked with the user level application. As illustrated in A windowed logging protocol implemented in the checkpoint library commits (i.e., inserts) each message mi of the i messages sent by a process Ps to a local log before passing mi to the OS for transmission. Messages are uncommitted (i.e., removed) from the log by the checkpoint library interposed with the receiving process Pr before they are passed up to the layers above for computation. One of skill in the art will recognize that the logging method as described in the embodiments herein is implemented as an online protocol (i.e., the protocol requires no knowledge of future inputs). However, it is possible that in other embodiments a non-online protocol might be used. In this embodiment, this uncommitting may take the form of a second small message mia from Pr to Ps that confirms successful delivery of mi. In the event of failure and subsequent restart the checkpoint library interposed with each process resends or �replays� each mi that was committed in the log when the checkpoint was taken. Should any mi have already been received by Pr before the checkpoint, it is discarded without notification to the layers above and also is uncommitted from Ps's log. Had mi not been received it is processed as a normal message. The windowed logging protocol precludes any inconsistency in the state of the communication channel. Note that this embodiment differs significantly from traditional message logging in that only messages in the log are replayed. The message log only contains messages which have not been received by the intended receiver. All messages that were successfully received by the receiver and acknowledged as such are uncommitted from the message log. During the course of execution, all dynamic memory allocations and subsequent memory modifications may be recorded by the checkpoint library through the use of a double-buffered commit list consisting of a front buffer and a back buffer. When the state of the application process is to be recorded, the front and back buffers are swapped and all modified memory regions are protected from modification. A separate thread of execution (i.e., the state recording thread) is spawned to record all of the modified memory regions listed in the previous commit list to non-volatile storage such as in shared secondary storage facility 110 (see Note that in this embodiment, there is no involvement of local secondary storage 114 (see If the application process attempts to modify a region of memory listed in the previous commit list it will be detected by the state recording thread. The thread will interrupt its in-order traversal and record the region in question so as to allow computation to proceed. The region will be listed in the current commit list to ensure the new modification is recorded in the next checkpoint. This non-blocking, copy-on-write, out-of-order local checkpoint commit method reduces a checkpoint from a global synchronization operation to a local operation. It transforms the commit operation to one that is bounded by the OS memory protection mechanism as opposed to one bounded by the write access speed of the secondary storage system. The two novel mechanisms above are described with respect to an illustrative embodiment to present a new method with which to coordinate the checkpoint of a distributed computation in such a manner that ensures a globally consistent state. This novel method is: non-blocking, transparent at all levels of the distributed computation, and performed in such a manner that processes may be migrated to any node. The coordination can be initiated by any process through the broadcast of a �checkpoint command� to all other processes that comprise the distributed computation. A numerical id may be used to identify the interval of computation or �epoch� whose state is to be recorded. Each individual process then halts interprocess communication, commits its state (marked by the epoch id) and enters a barrier on interprocess communication. After exiting the barrier, interprocess communication resumes as part of the new interval. As recognized by one of skill in the art, a barrier is a way to synchronize the different processes taking part in a distributed computation and is a parallel operation used to hold a process until other processes taking part in the barrier have reached the barrier. The method discussed above combined with a distributed system constructed of individual nodes may provide a supercomputing resource with the same reliability as a traditional custom supercomputer. It may allow individual processes to be migrated from a failed compute node to a properly functioning compute node in a manner that allows a distributed computation to proceed unimpeded by the mechanical or software failure of any individual nodes that comprise a distributed system. It may provide this reliability in a manner that is non-blocking and completely transparent to all levels of the distributed computation as opposed to conventional methods that require computation to halt, modification of some level of the distributed computation (e.g., user application, communication middleware, and OS) and manual intervention to restart a failed computation. In one embodiment a runtime system provides several key enabling technologies to improve the reliability of and enable dynamic control over the execution of a distributed system. In particular, some of the main features of this embodiment are:
An embodiment will be illustrated below in an exemplary distributed computing system. It should be understood that, however, that the system and method described herein is not limited to use with any particular type of distributed system, but is instead more generally applicable to any distributed computing environment or application in which it is desirable to improve reliability to the levels experienced in a traditional custom supercomputing environment. As used herein, the term �distributed system� is intended to include any system which includes two or more components, either computers, machines or other types of processors. Each computer in a distributed system may be, for example, a Symmetric Multiprocessor (SMP) and contain multiple processors. The term �distributed computation� is intended to include any instance of computation that is comprised of two or more processes working in concert to accomplish a computational task. The term �process� as used herein is intended to include any type of program, instruction, code, or the like which runs on one or more computers or other types of processors in a distributed system. The processes that comprise a distributed computation may cooperate either through the explicit exchange of messages over an interconnection network, the access and modification of memory regions that are shared by all processes, or some combination thereof. In the present embodiment all processes execute concurrently on distinct separate processors and each process will be illustrated as an OS process. The system and method discussed herein is not limited to such an environment however, and may be utilized regardless of the manner in which instances of computation are realized (e.g., user level threads, kernel level threads, and OS process). As used herein, the term �local checkpoint� is used herein to describe the recording of the set of information that fully characterizes the state of a single process contributing to the execution of a distributed computation. The term �checkpoint� is used herein to describe the aggregate set of local checkpoints which represents a globally consistent state of a distributed computation. The term �checkpoint interval� refers to the period of normal computation and communication that takes place between two checkpoint operations. The term �head node� refers to a special component in a distributed system on which distributed computations are submitted by users for execution and on which may run central monitoring and resource administration software. The term �compute node� refers to a single node in a distributed system that contains n processors (where n≧1) and therefore may host n processes. When a distributed computation is initiated on a conventional distributed system the execution of the individual processes starts on a subset of the compute nodes. Should a hardware or system software failure occur on any one of those compute nodes, the distributed computation halts. Most conventional distributed systems provide no checkpointing mechanism, and the distributed computation must be restarted from the initial point of execution. Should the distributed system utilize a distributed snapshots based mechanism to checkpoint distributed computations, the failed node must be repaired through manual examination and maintenance prior to the restart of the distributed computation. Even in the presence of sophisticated system software to automate the monitoring and maintenance of a compute node, hardware failures still necessitate the manual intervention of a system administrator for a non-trivial number of node failures. The distributed computation resumes at the state of execution recorded by the most recent checkpoint, on the exact subset of compute nodes that it was originally running on. Local checkpoints taken by a distributed snapshots based mechanism flush communication channels without capturing any of the state maintained in the OS and middleware and hence do not allow migration. Open communication channels depend on sockets or some other equivalent system level interface that are tied to specific network or hardware level addresses (e.g., IP addresses or Infiniband connection state). Further inspection of the requirements reveals significant drawbacks in the distributed snapshots based checkpointing mechanism for the majority of conventional distributed systems. Most distributed systems actually execute a number of distributed computations concurrently, each utilizing a subset of the compute nodes. Should one of the compute nodes fail, the distributed computation it was taking part in halts and the remaining, properly functioning, compute nodes in the subset become idle. In a distributed snapshots based checkpointing system, these compute nodes must remain idle until the failed node is repaired so that the distributed computation may resume execution on the exact subset of compute nodes it was originally running on. The amount of time that the non-failed nodes in the subset remain idle results in a waste of resources, adversely effecting the efficiency of the distributed system. The alternative is to allow the non-failed compute nodes to take part in other distributed computations, while waiting for the failed node to be serviced. Unfortunately, there is no upper bound on the amount of time it may take for the now fragmented original subset of nodes to simultaneously become available to resume the failed distributed computation. In such an environment there is no guarantee failed distributed computations will ever complete. In contrast, the present embodiment may take advantage of the high probability that at any given time in a distributed system a small number of compute nodes will be idle for lack of a combination of distributed computations that precisely utilizes all of the compute nodes. As a result, if one of the compute nodes actively taking part in a distributed computation should fail, the process(es) it hosted can be migrated to one of the idle �hot-spare� compute nodes. Additionally a system administrator may explicitly reserve a set of compute nodes to ensure that hot-spares are always available. The distributed computation is then able to continue unimpeded. Should the distributed system be used to execute a single distributed computation that consumes the entire set of compute nodes, it is left to the administrator to reserve a handful of nodes for fail-over purposes. It can be shown that the efficiency lost through this method is far less than the loss due to failures in a conventional distributed system. This embodiment provides a method for detecting the global state of a distributed computation and transparently acquiring local checkpoints of each process taking part in the computation in a manner such that the aggregate is globally consistent and local checkpoints are migratable. The embodiment may record local checkpoints with a novel out-of-order copy-on-write mechanism that allows computation to proceed during the course of a local checkpoint. The embodiment may be configured such that a distributed computation may be checkpointed and have some subset of its processes migrated for any or all of the following reasons: (1) a compute node failure occurs, either due to hardware failure, system software error, or any other itinerant failure that interrupts the process(es) executing upon it; (2) a system administrator wishes to remove a number of compute nodes from the distributed system for maintenance or upgrades without disrupting the distributed computations taking place; (3) an intelligent scheduling entity, managing the resources of the distributed system and utilizing a preemptive scheduling algorithm, wishes to temporarily halt a distributed execution to allow its compute nodes to be used for another computation; or (4) application software written with advance knowledge of the embodiment specifies a checkpoint at a point in execution (e.g., end of a time step) where the amount of state required to be saved is minimal (e.g., a user-directed checkpoint). When a distributed application is to be preempted, each instance may be sent a signal that is asynchronous to the computation in progress. This signal may be either a true OS signal or an out-of-band message. On receiving the signal, the checkpoint operation is initiated. Once the checkpoint operation has completed, a scheduling entity (executing a preemptive scheduling algorithm) may use OS facilities to halt or terminate the distributed computation. The scheduling entity is then able to instantiate another computation on the compute nodes that have been idled. Interconnection network 102 may be, for example, an Internet-based network. One or more processes 120 may be executed on each compute node 104. For example, a process P1 may run on compute node C1, and a process Pn may run on compute node Cn. Each process 120 may be executed, for example, by one or more processors (e.g., an INTEL ITANIUM microprocessor) included in compute node 104. The compute nodes 104 in the system are also connected to a shared secondary storage facility 110 (designated by S). The shared secondary storage facility 110 may be, for example, an NFS file system available over interconnection network 102 or a more advanced parallel file system such as the LUSTRE system. With respect to secondary storage facility 110, the same file system should be visible to any of the compute nodes 104 that are to be migration targets. In a typical embodiment, shared secondary storage facility 110 is accessible by all compute nodes 104. Each compute node 104 may include local memory 112 (e.g., dynamic RAM), which may be used, for example, to store user-level applications, communications middleware and an operating system, and may also include local secondary storage device 114 (e.g., a hard drive). Local memory 112 may also be used to store messages, buffers and logs associated with the checkpoint method described herein, as discussed in more detail below. Head node 106 may also include local memory 116 and local secondary storage 118. The compute nodes C1,C2, . . . ,Cn may be computers, workstations, or other types of processors, as well as various combinations thereof. It will be assumed for purposes of illustration that in the system requests for distributed computations will arrive at H and be dispatched to an idle subset of the compute nodes. In such a system not using the novel methods described herein, if there does not exist an idle subset of the compute nodes large enough to satisfy the request, it will be queued by the resource management framework on H for execution at a later point in time when such a subset can be found. The resource management framework may attempt to reorder the queued request to minimize the number of compute nodes that are idle, but once a distributed computation has been launched it generally must be allowed to run to completion. It should be noted that the distributed computations are typically non-interactive. Referring to Checkpoint library 202 is illustrated in The windowed logging protocol may be used to ensure that the state of a communication channel and the two processes at either end remains consistent during a checkpoint. This approach is not obvious when considering the assumptions made by those using conventional distributed snapshots based systems. The distributed snapshots method assumes that all communication channels are reliable FIFO queues and so drains them of any state prior to checkpointing. Conventional methods thus do not record the messaging state tied to the reliable communication framework provided by the OS, including hardware specific addresses and current connection buffers, thereby preventing migration. In contrast, the present embodiment uses a general operating assumption that all communication channels provided by the OS are inherently unreliable (e.g., providing no guarantee of delivery or correctness of order). As stated above, all communication operations invoked by either the application or middleware may be directed to the checkpoint library. These operations are carried out utilizing the communication primitives provided by the OS. As OS channels are assumed to be unreliable, the present embodiment may ensure correct delivery of messages through use of the windowed message logging protocol. Upon receipt of the message the receiver Pr replies with a short acknowledgment message to inform Pr of successful delivery. This action taken by Pr uncommits the message from Ps's log. A unique monotonically increasing sequence number may be associated with each message so that messages received in the wrong sequence may be re-ordered properly by Pr. The sequence numbers also allow a receiver to detect that a message was lost (e.g., a gap in sequence numbers of messages received), as well as the receipt of duplicate messages. Duplicate messages are discarded (since they were already received and processed), while lost messages may be dealt with by requesting their retransmission. Messages that are listed in the sender log are known as �outstanding� messages. In order to permit the use of a finite buffer for the log, a limit may be placed on the number of outstanding messages that are allowed at any one time. Should this limit be reached, Pswill cease sending messages until the oldest outstanding message is acknowledged. The range of sequence numbers that is allowed to be outstanding is known as the �transmission window�, and as messages are sent and received the transmission window in essence slides over the total set of messages that the application and middleware request to be transmitted. The present embodiment implements communication in the manner described above and, as mentioned, assumes that the communication channel which starts at Ps's interface to the OS and ends at Pr's corresponding interface is unreliable. It should be noted that in such an environment, it is generally not possible to determine whether the loss, out-of-order delivery, or duplication of a message is due to a failure of the OS, the failure of the interconnection network, the failure of the OS on the node at the remote end of the communication channel, or even the failure of the process executing on that remote end. Accordingly, the present embodiment does not checkpoint state within the OS or in-flight over the communication fabric. Nor does the present embodiment engage in coordinated �flushing� of messages out of the OS and communication fabric before checkpointing as do conventional systems. Rather, the windowed message logging implemented by the user-level library of this embodiment masks any loss of state within the operating system and the interconnection network during the checkpoint operation as a message loss that would be dealt with in the normal flow of communication over an unreliable channel. As discussed above, a deterring factor in aggregating the local checkpoints from Ps and receiver at either end of a communication channel into a globally consistent checkpoint is an inherent difficulty in preventing inconsistencies in the communication channel state from occurring upon restart. In a conventional distributed system the inconsistencies that may arise when checkpointing while Ps is sending a message m to Pr are as follows:
1 3. As the embodiment also adds a secondary acknowledgment message ma from Pr to Ps, there is a third possible inconsistency when Pr checkpoints after sending ma, whilst Ps checkpoints before receiving the acknowledgment. Use of the novel windowed logging method described herein may avoid the above failure modes. In the first failure mode, Ps is restarted in a state prior to m having been sent and sends m as part of the normal flow of execution. Pr, having been restarted in a state where m has already been received, receives m again, resulting in an inconsistent global state. The windowed logging protocol may prevent such an occurrence by recognizing from the sequence number associated with m that the message is a duplicate and discarding it. The receiver then uncommits m from Ps's message log by replying with ma. In the second failure mode, when Ps is restarted in a state after having sent m and Pr is restarted in a state prior to the receipt of m, the message m will never be received by Pr. The windowed logging protocol prevents the permanent loss of messages. As m is never received, Pr will never send ma to uncommit m from Ps's log. After a certain period of time t has elapsed, m will �timeout� at Ps and be resent. Pr will now receive the second transmission of m and uncommit it from Ps's log by replying with ma. An appropriate value for t is set based on the bandwidth and latency of the specific interconnection network utilized by the distributed system. In the third failure mode, when Ps is restarted in a state prior to having received ma and Pr is restarted in a state after ma has been transmitted, m will never be uncommitted from Ps's log. Such an occurrence may be prevented through the use of a windowed logging protocol. After a certain period of time t has passed, m will �timeout� at Ps and be resent. Pr will now receive the second transmission of m, recognize from the sequence number that it is a duplicate, and send a second copy of ma, thereby uncommiting it from Ps's log. The windowed message logging protocol may be used to provide another fundamental advantage over conventional distributed snapshots systems. The present embodiment provides a distributed system with the ability to transparently migrate an individual process to a different compute node. As stated above the conventional distributed snapshots method prevents inconsistencies by using marker messages to flush the network of all messages sent prior to checkpointing. While this approach results in a globally consistent checkpoint, it does not capture any of the state maintained by the communication middleware or OS, including but not limited to open connections bound to specific hardware addresses. In any type of transparent distributed snapshots system this precludes any type of migration, as any remote process will attempt to communicate with the migrated process using the address of the original node it was executing on. Some distributed snapshots systems attempt to work around the problem by leaving a small �forwarding� process Pf on the original node. Pf receives all traffic at the original node intended for the migrated process and forwards it to the new location. However, this model does not address fault tolerance as any failure precludes the execution of Pf. In order for a distributed snapshots based system to provide migration for fault tolerance, it must modify either the communication middleware or the OS. A prominent example of such a system is the communication middleware (e.g., LAM/MPI) which has special code to implement a distributed snapshots protocol and integrate with a third party single process checkpoint system. This type of arrangement is undesirable as it ties user applications to a specific implementation of a messaging-passing API, thereby defeating the primary motivation for a general message-passing specification. OS modification is even more unfeasible as many distributed systems run general purpose OS's that would require manual modification by system administrators. It is generally desired that all facets of checkpoint/restart, including migration, are be transparently handled by a stand-alone framework, requiring no modification of user applications, middleware, or operating systems. The disclosure herein may permit the providing of such transparent migration as a benefit of the windowed message logging protocol above. All inter-process communication is redirected through the checkpoint library in order to implement the logging protocol. As the embodiment handles all communication, all that is required to enable migration is redirect all communication intended for a given process to the node to which it has been migrated. To ease the implementation of this method, the present embodiment may virtualize network addresses, meaning that all application and middleware requests for OS communication interfaces are satisfied with addresses that the present embodiment provides. Internally, these addresses may be mapped to the actual hardware addresses of the compute nodes. When a migration occurs, the checkpoint library modifies the mapping accordingly. In such a manner, the present embodiment is able to provide transparent support for process migration. In order to locally checkpoint a process the method records the following: (1) a process's global and static data, (2) its processor context (contents of the registers), (3) its stack, and (4) all dynamically allocated memory. The first three components are of a known size and can be recorded by simply writing their values to stable storage. The last component, dynamically allocated memory, may refer to both memory allocated out of a common area (e.g., the heap), or memory that is �mapped� in regions whose size is a multiple of a unit size (usually an operating system page). Although it is possible to simply record all the dynamically allocated memory regions with each checkpoint, the typical distributed computation utilizes large amounts of memory and would require substantial amounts of disk access with each checkpoint. It is more desirable to only checkpoint the memory regions that have been modified in the interval since the last checkpoint, creating an ordered set of �incremental� checkpoints. To achieve this, the checkpoint library may be transparently interposed in the control path of all allocations. To determine which memory regions need to be stored as part of a local checkpoint, the present embodiment utilizes an incremental change detection method similar to that, for example, found in James S. Plank, Micah Beck, Gerry Kingsley, and Kai Li. Libckpt: Transparent Checkpointing Under Unix. Technical Report, 1994, which is incorporated herein by reference, or similar to that, for example, of other conventional single process checkpointing methods. All dynamically allocated memory is write protected and any attempts by the application or middleware to modify protected memory results in a segmentation violation. The signal handling of the process is interposed to redirect signal notification of segmentation violations to the checkpoint library. The checkpoint library determines if the segmentation violation was due to a modification of write-protected memory or an actual segmentation violation incurred by the application or middleware. True segmentation violations are passed to the system error handling routines. If the violation was due to a modification of write-protected memory, the region of memory that contains the modification is marked in a commit list as having been modified. The write protection is removed and the application's normal flow of execution proceeds with all future modifications to the region ignored until the next checkpoint. A region in the commit list is checkpointed in the presence of any modification regardless of the number of modifications in a checkpoint interval. When the time comes to take a checkpoint, computation is suspended and the commit list is traversed in order, and all of the regions of memory that have been modified in the interval since the last checkpoint are recorded. The suspension of computation prevents regions in the commit list from being changed during their storage into non-volatile memory. In this incremental checkpointing approach, the present embodiment determines the minimal amount of dynamically allocated memory that must be recorded. The size of a region of memory that is marked is, for example, a discrete multiple of the smallest sized region of memory that the OS's memory protection mechanism will operate on. On many OS's this is, for example, a page. Note that the granularity of the memory region size (e.g., one or a multiple of pages) may be varied in order to improve performance. The incremental checkpoints may be, for example, stored as a series of files, or as a set of incremental changes to a single double-buffered checkpoint file. An advantage of a double-buffered file is that a failure in the middle of a checkpoint operation will not result in a corruption of previous checkpoints. This suspension of computation becomes problematic when considering that incremental checkpointing may require that the amount of memory to be recorded for each checkpoint may be large, sometimes requiring all dynamically allocated memory to be stored in every checkpoint. This pathological case can be common in the HPC applications that distributed systems are often used to execute. For instance, in applications that solve large systems of linear equations using matrices, the matrices are split into pieces (e.g., sub-matrices, columns, and rows) and each process operates on a piece. Many common matrix operations modify the value of every item in the matrix, and as such would require the entire matrix to be checkpointed. Factoring the relative speeds of the processor and access to the non-volatile storage subsystem, processors in a distributed system can spend significant amounts of time idling, waiting for checkpoints to be written to storage. This blocking problem is exacerbated as the distributed computation will not resume until every node has finished storing the memory regions listed in its commit list and resumes local computation. These limitations transform the checkpoint operation of a conventional system into a de facto global synchronization whose lower bound is determined by the slowest process. To reduce the operation of committing a checkpoint from a global synchronization operation to a non-blocking local operation whose runtime is not bounded by the access speed of the non-volatile storage system, this embodiment may make use of a novel non-blocking, copy-on-write, out-of-order checkpointing approach described by the flow diagram of Note that although the embodiment described below discusses the swap as taking place prior to memory protection, in other embodiments the swap could be done after write-protecting the memory regions while they are located Lc. The buffer, whether Lc or Lp, that contains the regions to be protected depends on when during the freeze period the buffers are swapped. If the swap occurs before write-protecting, the regions are found in Lp during the protection phase, as discussed in the embodiment below. The write protection is performed prior to exiting the freeze period. The double buffer itself may be implemented with any suitable data structure such as, for example, a linked list. The swap is implemented quickly by recording the address in memory that the two buffers are located at and simply exchanging the two addresses when we want to �swap� the buffers. A second thread of execution is spawned off to begin recording the contents of all memory regions listed in Lp and the application is allowed to resume computation interleaved (i.e., occurring in parallel) with the checkpoint storage operation. After a memory region is recorded, its listing is removed from Lp. Should the interleaved computation attempt to modify a memory region, that modification will be detected, and the present embodiment will determine if it is a region listed in Lp. If the region is found listed in Lp, the recording thread's traversal of Lp is interrupted to perform an out-of-order recording of the memory region to secondary storage. This ensures that the checkpoint that was in the process of being recorded remains consistent. The memory region is added to Lc, as the modification in question will need to be recorded in the next checkpoint. Write-access to the region is restored and application execution allowed to proceed. With this method the global consistency of a checkpoint may be assured, while allowing the interleaving of application execution with checkpoint I/O operations. It should be noted that the manner of data structure used to represent the buffers is irrelevant, and generally differentiates only in the efficiency with which the memory regions are listed and traversed. Note that the regions to be write-protected are found in the region double buffer Lp/Lc. The double buffer does not need to be write-protected itself since the double buffer is allocated and maintained by the checkpointing library. The memory regions containing the logs for the windowed message logging are write-protected and will be listed in the double buffer. An advantage of this novel approach is that the suspension of the application is now a local operation and the checkpoint phase no longer requires a global synchronization. The second fundamental advantage to this approach is that the duration of the suspension is relatively short, since it only includes the time required to write protect the memory regions as opposed to recording the actual contents of the modified memory regions, an operation whose duration has a lower bound determined by the access speed of the non-volatile storage system, which can be orders of magnitudes slower than the speed of computation. The out-of-order copy-on-write mechanism prevents any of the regions listed in Lp from being modified prior to being recorded. The write protection in effect becomes the �commit� operation rather than the actual recording of memory regions to, for example, a hard disk. The embodiment discussed herein may be used to assist in circumventing problems that may arise when utilizing conventional distributed checkpointing systems and provides a novel non-blocking global checkpoint coordination method that is generally not possible in conventional systems. After one of the intervals has elapsed, some process Proot will initiate the checkpoint operation by broadcasting a checkpoint command to all processes taking part in the distributed computation. Proot may be any one of the processes taking part in the computation (e.g., a user-directed checkpoint) or another process executed on head node 106 such as, for example, a scheduling entity or resource management framework software (e.g., LSF, PBS Pro, and Sun Grid Engine). The broadcast command will contain the epoch identifier associated with the current checkpoint operation. Upon receiving the broadcast, each process Pi will enter a �freeze� period. During the freeze period, all inter-process communication and computation is suspended. The windowed message logging protocol ensures that any messages discarded during the freeze are not lost. After freezing, Pi will utilize the double-buffered commit list and swap the contents of Lp and Lc, before write-protecting the memory regions listed in Lp. The out-of-order copy-on-write ensures that Pi is now able to consider its state from the previous interval committed and spawn off the storage thread of execution. Pi exits the checkpoint operation by entering an out-of-band barrier. The barrier operation is expected to normally occur quickly, since all processes of the distributed system involved in the computation typically receive their checkpoint broadcast separated by no more than the interconnection network latency, and hence enter the barrier separated by a delay usually bounded by the network latency. In order to implement the barrier, each process Pi will notify the initiating process Proot out-of-band that Pi's state has been committed. When Proot receives a commit confirmation from each Pi included in the computation, it will transmit an out-of-band broadcast declaring the interval finished. Upon receipt of this broadcast, Pi exits the barrier, the freeze is lifted, and Pi resumes inter-process communication and computation. Note that the barrier is not necessary for the correct operation of this embodiment. The barrier merely ensures that all processes involved in the distributed computation have received and processed the checkpoint message, in particular that all processes are operating at the same epoch. The �current epoch� identifier is a monotonically increasing 64-bit quantity of which each process maintains its own local copy (e.g., as a value in local memory). Each process increments this local copy after exiting the global checkpoint operation and prior to resuming local computation and interprocess communication. The incremental checkpoint files may be ordered and identified using this epoch identifier. The only time at which two processes may have different values for the current epoch is during the global checkpoint operation when processes that have exited the operation have a current epoch that is one greater than the processes still taking part in the global checkpoint operation. If for any reason at any time during a checkpoint interval any two processes in the distributed computation are operating at different epochs, this fact may be detected as a failure. The process which detects this failure will broadcast an exception, and the distributed computation may be rolled back to its most recently known globally consistent checkpoint. The presence of the barrier merely accelerates the detection of this failure mode. The above embodiment's entire checkpoint phase may be only loosely coordinated, and it typically may last only a fraction of the period of time necessary to commit the actual contents of modified memory to secondary storage. Now briefly discussing some concepts regarding global consistency that are believed to be applicable to the embodiment described above, consider a distributed computation comprised of n processes. A global state of the system can be abstractly defined as the union of the individual state of each process Pi and the state of the communication channels. For the purposes of this brief discussion, it is assumed that the distributed system is a message passing system. However, it should be noted that the discussion is also applicable to shared memory systems where communication between processes is facilitated through the modification of shared memory locations. For a distributed computation running on such a system, the global distributed state SG can be more formally defined as:
The role of each process Pi in the global checkpoint operation has been reduced to executing a local checkpoint operation to record Si. In order to ensure the consistency of SG, no state Si may change during the local checkpoint operation. More specifically upon entering the global checkpoint operation no process Pi may change its local state or the state of any other process Pj until the global checkpoint operation is finished. The only self-inflicted cause of local state change is local computation. Likewise the only manner for Pi to change the state of Pj is to send a message. Given these criteria, recall that upon entering the global checkpoint process, Pi stops local computation and enters a �freeze period� during which all interprocess communication is suspended. Pi then executes the local checkpoint operation and exits the global checkpoint operation by entering the loosely synchronized out-of-band barrier operation. At no point during its part in the global checkpoint operation does Pi alter its own state or send a message to any process Pj that would alter Sj. While in the loosely synchronized barrier operation Pi resumes local computation but still refrains from any inter-process communication. This ensures that Pi does not alter the state of any process Pj that may still be taking part in the global checkpoint operation. Only after every process enters the loosely synchronized barrier, which subsequently terminates, does Pi resume interprocess communication. Using the foregoing approach provides the global consistency of all global checkpoints taken. The embodiment described above may be used to implement a preemptive scheduling system capable of providing maximum flexibility when implementing a resource sharing policy. The preemptive scheduler preferably should use a user-level transparent windowed message logging based checkpointing framework such as described above to record the state of executing jobs. When the resource sharing policy dictates that one computation is to be suspended and another started or resumed in its place, the embodiment described above may be used to checkpoint the executing computation before halting it. A second computation that may be a new computation or one resumed from a checkpoint will be executed. By using a transparent windowed message logging based method, migration of individual processes may be permitted and no restrictions are placed on the physical location of processes in the distributed system. This enables the scheduler to preempt and resume computations according to a resource sharing policy bounded, for example, only by the total capacity of the distributed system. The preemptive scheduler may also take a relative priority of each computation into account by varying the length of an individual computation's interval of execution. The preemptive scheduler is generally not required to account for the spatial locality of processes being preempted and resumed, although it should be noted that the spatial locality may be configured to minimize the latency of the communication channels used by the cooperating processes. A specific example of one approach is described by the following steps: Execute a monitoring process or daemon on each compute node in a distributed system. Execute on the head node or one of the compute nodes of the distributed system a scheduling process that allows users to queue distributed computations for execution on the distributed system. Link applications against a user-level checkpointing library that is transparent to the application, communication middleware and operating system and that virtualizes network/hardware addresses using a mapping of virtual to network/hardware addresses. Initiate a distributed computation by sending a message to the daemon residing on each node being utilized in the distributed computation. The daemon will initiate a new operating system process to begin execution of the application. After a period of computation, broadcast a checkpoint message to the monitoring daemon on each compute node taking part in the computation. The daemon will induce the process to checkpoint its state. This inducement may take the form of, but is not limited to, an operating system signal, a modification to a shared memory location, or a message sent over an out-of-band communication channel. Checkpoint the process with a method that uses a windowed message logging protocol to ensure that the set of local checkpoints results in a consistent global state and that the suspended process may be migrated to any other node in the system. After all the daemons have sent notification of a successful checkpoint back to the scheduling process, broadcast a kill message to the monitoring daemon on every compute node taking part in the computation. Each daemon will receive the message and induce the process to halt, using a method similar to those listed above. After the distributed computation has halted, another computation may be either started or resumed from a previously-saved checkpoint on the set of now idle nodes. When resuming a computation, the daemon will provide the process being resumed with any changes made in the virtual to network/hardware address mapping as a result of process migration. This information may be provided in the form of, but is not limited to: a flat text file, shared memory locations, or environment variables. After a period of execution this second computation may itself be preempted in the manner described above and replaced with the first computation or some third computation. In the presence of multiple-queued computations, migrate resumed computations to the subset of nodes that best facilitates minimal idle CPU time while best approximating the resource sharing policy in place. By periodically preempting and/or migrating computations in an ordered manner, a distributed system may concurrently execute any number of distributed computations such that the elapsed time from when a task is submitted to the queue to when it completes is bounded only by the runtime of the computation and the number of tasks in the queue, and will not be affected by the order in which it was queued. If prioritized resource usage is desired, assign each task a priority value when it is queued. Preempt executing tasks when higher priority tasks are added to the queue. Finally, adjust the period of time in between preemptions such that higher priority tasks are accorded the desired percentage of compute cycles. Although a specific exemplary method was described in the preemptive scheduling steps presented above, one of skill in the art will recognize that in other embodiments many of the above steps may be re-arranged and/or omitted. By the foregoing disclosure, an improved system and method is described that may be used for handling of the checkpoint/restart problem in a manner that is non-blocking, that may enable the migration of individual processes, and that may be transparent to the application, any middleware that is in use, and the operating system. Other advantages may include one or more of the following features:
The foregoing description of specific embodiments reveals the general nature of the disclosure sufficiently that others can, by applying current knowledge, readily modify and/or adapt it for various applications without departing from the generic concept. For example, although the disclosure was presented above in the context of using a checkpoint library as a user-level library, in certain other embodiments the checkpoint library could be implemented as a kernel module. Therefore, such adaptations and modifications are within the meaning and range of equivalents of the disclosed embodiments. The phraseology or terminology employed herein is for the purpose of description and not of limitation. Referenced by
Legal Events
Rotate |

subset: commoncrawl


text:
True commitment to
a feeling, idea, cause, person -
what is it?
Civility? Conflict? Mom?
Honor? Pudding?
What measurement used,
what's his worth…
Did he use his life?
His talents, detriments?
Who remembers him?
Is it so vital
if he made a difference?
His one sole reason for being,
the purpose God wrought him for
was to fill a space in the crowd.
Ever think about that, Superman?
- 1
- 2
- 3
- 4
- 5
- Recent
Comments There are no recent comments (9 older comments) - Click here to add a comment to this poem or Show more comments or Read All User Comments (9)
- Add a
Comment - Send
Feedback Send private anonymous feedback to the author (click here to post a public comment instead).


text:
Power Tools & Accessories
Tool Shed 29 Piece Titanium Drill Bit Set PS07350
$19.99The Tool Shed 29 Piece Drill Bit Set includes 29 popular size bits for drilling into wood, metal and plastic. Hang photos or drywall, or install shelving for storage or new hardwood floors. The drill bits are titanium nitride coated for durability. Use with a cordless drill for DIY, home improvement and general maintenance projects around the house. You will find no end of uses for this Drill Bit Set. Packaged inside a storage case to keep the bits organized and easy to find.
29 Titanium Drill Bits:
- 1/16"
- 5/64"
- 3/32"
- 7/64"
- 1/8"
- 9/64"
- 5/32"
- 11/64"
- 3/16"
- 13/64"
- 7/32"
- 15/64"
- 1/4"
- 17/64"
- 9/32"
- 19/64"
- 5/16"
- 21/64"
- 11/32"
- 23/64"
- 3/8"
- 25/64"
- 13/32"
- 27/64"
- 7/16"
- 29/64"
- 15/32"
- 32/64"
- 1/2"
Dewalt 6" x .040" x 7/8" Metal and Stainless Cutting Wheel DW8725
- Thin .045" wheel design for fast burr free cutting
- Proprietary aluminum oxide grain combination for aggressive cutting action
- Proprietary material mix ensures durable long life wheels
- 2 full sheets of fiberglass for durability and safety
Milwaukee M18 Cordless Lithium-Ion Compact Driver & 1/4" Hex Impact Diver Combo Kit 2691-22
$199.99Since 1924, industry-leading Milwaukee Electric Tool Corp. has been producing the highest-quality, most-reliable, heavy-duty electric power tools and power tool accessories. This trusted leader brings you the ultimate M18 cordless 2-tool combo kit (2691-22). This set includes the Compact Driver (MWK-2601-20) and the 1/4" Hex Impact Driver (MWK-2650-20). Patented design, state-of-the-art 4-pole frameless motor, and user-friendly ergonomics make the M18 Cordless System an impressive series that maximizes both power and performance. Milwaukee has taken no shortcuts in producing this power kit. Each tool also features variable speed triggers, built-in LED illumination, battery fuel level gauges, and lightweight yet durable construction.
Milwaukee knows job sites can be brutal on a tool, so they have carefully designed these battery packs to be rugged and durable. Features include cell separators, welded frame, and recessed terminals. Over-molded housing and slide pack dual clips round out each battery pack's practical construction. A 30-minute battery charger completes the kit.
Innovative features, unmatched power, and exceptional runtime make the Milwaukee M18 Cordless Lithium-Ion Compact Driver and ¼" Hex Impact Driver Combo Kit essential for any job site. Arsenal comes complete with a sturdy carrying case and a handy belt clip.
Milwaukee M18 Lithium-Ion Compact Drill Driver
Rather than drilling pilot holes for ductwork or running wire and conduit, the Milwaukee M18 lithium-ion compact drill driver is the best and most powerful choice. With 24 clutch settings and delivering 400 in/lb. of torque, this driver is simply the most powerful in its class. Adjustable speeds range from a low speed RPM of 1-450 to the high speed RPM of 1-1,800. Weighing 5.3 pounds, measuring 5 ¾ inches, and featuring a convenient side handle, the M18 drill driver is both comfortable and compact. And with exceptional drill bit retention, it is easy to see why this driver is an easy choice for at-home handymen and professional builders alike.
Milwaukee's M18 ¼" Hex Compact Impact Driver
When you need to secure sheet metal screws or 5/8-inch and 3/4-inch pipe couplings or any other hex fastener, reach for Milwaukee's M18 ¼" Hex Compact Impact Driver. Though it weighs only 3.5 pounds, it packs major power and delivers 1,400 in./lb of torque and 0-3,200 IPM. Featuring 0-2,200 RPM, the Milwaukee driver is over 70% faster than the competition. Though the fastest, strongest, and most durable, this impact driver also features user conveniences. It includes one-handed bit-loading and, like the drill driver, it also has LED illumination for those tough jobs with poor lighting. And with twice as much run-time, you can maximize the power and minimize the downtime.
Red Lithium Compact Battery Packs
Powered by Milwaukee's high-capacity Red Lithium Compact Battery Packs (included), these tools boast longer run times and are, therefore, much more user-friendly. In 2005, Milwaukee introduced the first-ever lithium-ion charging system. Since, they have perfected the formula and have become an industry leader. Their optimal batteries, though 50% smaller and 65% lighter, are 30% more powerful than NiCd compact batteries. The convenient gauge tells users when the charge is running low. Just one more feature that reduces your unproductive downtime.
- Two power tools: Compact Driver (2601-20) and ¼" Hex Impact Driver (2650-20)
- Carrying case
- Two batteries: 18V RED LITHIUM Compact Battery Packs
- 30-minute M18 Battery Charger
- Belt clip
Key Features:
- 18V voltage
- Built-in LED illumination
- Powerful torque
- Battery fuel level gauges
- Adjustable speeds
- 5-year manufacturer's warranty
Milwaukee M18 18 Volt 2.0 Amp Compact Red Lithium-Ion Battery 48-11-1820
Delivers up to 2X more runtime, 20% more power and 2X more recharges than standard lithium-ion batteries. The M18™ REDLITHIUM™ 2.0 Compact Battery Pack features superior pack construction, electronics and performance to deliver more work per charge and more work over pack life than any battery on the market. It offers 2.0 amp-hours of runtime. With Milwaukee durability built into each pack, this battery runs cooler and performs in climates below 0°F/-18°C. Its compact, lightweight design fits in tight work spaces and reduces user fatigue.
Managed by Milwaukee's exclusive REDLINK™ Intelligence, the battery features overload protection to prevent you from damaging your cordless power tools in heavy-duty situations, while the discharge protection prevents cell damage. The temperature management system and individual cell monitoring help maximize battery life. Switch to the REDLITHIUM 2.0 Compact Battery Pack to instantly upgrade the runtime and durability of your Milwaukee M18™ cordless tools.
- (1) M18™ Compact RED LITHIUM™ Battery
The 18V Compact LITHIUM-ION battery works on the following M18 tools:
- 2601 - M18 ½" Compact Drill/Driver
- 2650 - M-18 ¼" Hex Impact Driver
- 49-24-0171 - M18 Work Light
- 0880-20 - 18V Cordless Vacuum
- 2.0 Amp hour - 36 Watt hour battery
- LITHIUM-ION battery delivers long life and run time with fade-free power
- Battery fuel gauge - Displays remaining run-time
- Milwaukee REDLINK™ Intelligence
- Overload Protection - Prevents user from damaging their cordless power tool investment in abusive situations
- Temperature Management System - Keeps battery in ideal temperature range to provide maximum life
- Individual Cell Monitoring - Ensures optimal charge and discharge for maximum life
- Discharge Protection - Prevents cell damage from over-discharge
- Integrated weld frame
- Cell separator
- Impact and vibration protective rubber
- For use with Milwaukee M18™ cordless tools
- Voltage: 18v
- Battery Type: Lithium Ion
- Warranty: 2 Year
- Weight: 0.95 lbs
Dewalt 20V Max Cordless Lithium-Ion 1/2 in. Compact Drill Driver Kit DCD771C2
- Compact, lightweight design fits into tight areas
- High speed transmission delivers two speeds (0 - 450 / 0 - 1,500 RPM) for a range of fastening and drilling applications
- Ergonomic handle delivers comfort and control
- Heavy duty 1/2-in ratcheting chuck with carbide inserts provides superior bit gripping strength
- Two-speed, all-metal transmission for increased run time and longer tool life
Forney Industries Wheel Cut-off 4-1/32 X 3/8 71856
$1.194" X 1/32" X 3/8" Type 1 Metal cut off wheel (A60T) Max RPM 18,000. Learn More -
Dewalt 4 1/2" x 1/4" x 7/8" High Performance Metal Grinding Wheel DW4514
- High performance aluminum oxide grain for long life and high material removal
- High grain concentration for agressive grinding action and smoother running wheels
- Proprietary material mix ensures durable long life wheels
- 3 full sheets of fiberglass for durability and safety
Dewalt 18V XRP Battery Combo Pack DC9096-2
- (2) High capacity XRP™ batteries have 40% more run-time than standard batteries
- Powers entire DEWALT® line of 18V tools allowing users to run multiple tools off of the same battery pack
Irwin Marathon 2 1/2" Bi-Metal Hole Saw 373212BX
$9.99Product Features- Variable depth gullets provide more efficient chip clearance
- 4/6 Variable pitch teeth for reduced noise and improved performance
- Electron beam weld for stronger, longer lasting teeth
- Bi-metal construction for toughness and long life
Irwin Marathon 1 1/8" Bi-Metal Hole Saw 373118BX
$5.99Product Features- Variable depth gullets provide more efficient chip clearance
- 4/6 Variable pitch teeth for reduced noise and improved performance
- Electron beam weld for stronger, longer lasting teeth
- Bi-metal construction for toughness and long life


text:
K And G Fashion Deals
- Recent Deals
- Popular Deals
- On Sale 15 | 30 | 45 | 60
This set of peel and stick wall decals is guaranteed to delight Spider-Man and Marvel Heroes' youngest fans. Spider-Man, Wolverine, the Thing, the Lizard, the X-Men, the Incredible Hulk, and more are rendered in their cuddliest, most fun-loving fashion. Check out the adorable self-adhesive border (sold separately) to complement the look.
Advanced Filter


text:
Know more about this business than we do? Cool! Please submit any corrections or missing details you may have.
Help us make it rightWe Save Trees! Treat and Care for trees & Shrubs. Diagnose disease and other plant health problems. Perscribe and follow through with individualized treatment plans. Counduct damage & tree hazard assesments. TREE SPECIALISTS AND PATHOLOGIST
Be the first syracuse.com user to add photos or videos of Dr Daniel F Marion
Dr Marion and his son saved my sugar maple tress. I can't say enough good things about their care. I would definitely recommend them.
Tree Specialist/Pathologists
Tree Specialist & Pathologist
Hazard Tree Evaluations
Specialized Tree & Shrub Care Programs
Disease & Insect Control
Tree Damage
Tree Damage Care
Insurance Appraisals/Claims
Tree Specialist / Pathologist
Tree Preservation & Complete Treatment
Specializing In Tree Consultations & Appraisals for Insurance & Legal Claims
Regular Tree/Shrub Maintenance...
Fertilization, Tree Specialist/Pathologists, Tree Preservation & Complete Care, Laboratory Guided, Hazard Tree Evaluations, Disease & Insect Control, Diagnosis, We Save Trees, We Make Trees Healthy, Appraisals, Tree Specialists/Pathologist, Tree Specialist & Pathologist, Tree Specialist, Tree Preservation & Complete Treatment, Tree Injury Care, Tree Damage Care, Tree Damage, Programs for Tree &...


text:
Rhett Miller
Texas native Rhett Miller is perhaps best known as the frontman of the Dallas-based alt-country band Old 97's, although he also pursued a critically acclaimed solo career during the 2000s. Formed in 1993, the Old 97's built a devoted following with their blend of country and power-pop influences, which won the group a brief stay on the roster of Elektra Records. All four members also pursued side ...more


text:
HellHell, Norway
Related Reading
- Into the Forbidden Zone: A Trip Through Hell and High Water in Post-Earthquake Japan
- Holidays in Hell: In Which Our Intrepid Reporter Travels to the World's Worst Places and Asks, "What's Funny About This" (O'Rourke, P. J.)
- Hell or High Water: Surviving Tibet's Tsangpo River
- Home Below Hell's Canyon
- A Practical Guide to My China: How to Enjoy Traveling in Paradise and Avoid Travel Hell
Hell, Norway: It's lovely in Hell this time of year
Today is all about bragging rights: not everyone can say they've been to Hell and back, with the pictures to prove it.
This village, population 1500, has its fans (and Facebook page, of course) because, let's face it, it's irresistible to take pictures of yourself in Hell, especially in winter on the days Hell freezes over.
The railway in Hell (not Amtrak, surprisingly) has a large, curious sign on its front reading 'Gods-expedition'. Surely, with enough aquavit, you could come up with a plausible philosophical explanation of why God would want to go to Hell and would it still be Hell if God visited? Anyway, the facts are more prosaic: it's an old Norse spelling of cargo handling.
It's true that there isn't a whole lot to do in Hell, though if you visit from September 3-5 this year, Hell breaks loose with the Blues in Hell festival, disproving Shaw's contention that Hell is full of musical amateurs. In fact, you'll find good music and plenty of opportunity for more sophomoric wordplay.
For the record, the road to Hell is not paved with good intentions.


text:
'Midnight Rider' Producers, Gregg Allman Facing New Injury Lawsuit
Antonyia Verna says the February 20 train incident that killed one crew member left her with serious physical injuries and post-traumatic stress.
The hits keep coming for producers of Midnight Rider as makeup artist Antonyia Verna has become the latest to file a lawsuit over injuries connected to a train collision during the making of the movie about Gregg Allman.
In a Georgia county court, Verna has added her complaint to ones previously filed by the parents of Sarah Jones, a camera assistant killed in that Feb. 20 incident, and hairstylist Joyce Gilliard. Just like the past two lawsuits, Verna is suing Allman himself, as well as Unclaimed Freight Productions, Open Road Films, location manager Charles Baxter and other supervising personnel on the film.
Besides Jones' death, there have been reports of six other injuries at a property in Wayne County, Georgia, where a scene was to be filmed on railroad tracks on a trestle bridge. Verna, who previously showed up in an incident report released by the Wayne County sheriff's department as having been treated at the local hospital after the incident, is now looking to address her injuries.
According to her lawsuit, Midnight Rider producers didn't get permission to film on the trestle bridge and concealed that fact from the rest of the cast. The lawsuit then alleges that the defendants failed to take "minimum safety precautions" and failed to "comply with applicable industry standards."
This being a negligence claim, Verna's attorneys spell out the alleged lack of precautions taken by Midnight Rider filmmakers, including selecting an "unreasonably dangerous site for the filming location," failing to "station safety personnel," failing to station "look-out" individuals to watch for an approaching train, failing to "hold a safety meeting prior to filming," failing to "have an on-site medic present," and so forth.
Eventually, the train came at "rapid speed," and Verna's complaint says the crew was told they would have "60 seconds to remove themselves," but that amount of time wasn't enough for some of the crew running towards the train, as that path was apparently the shortest distance towards potential escape.
Verna says she suffered "serious physical injuries, medical and other necessary expenses, post-traumatic stress, mental anguish, lost income, and mental and physical pain and suffering."
She is demanding compensatory and punitive damages. She is joined in the action by her spouse Richard Brewer, who also wants monetary damages for his own suffering. They are represented by attorney Keenan Nix.
We'll add any responses from defendants. In the past, Allman and Open Road have questioned their inclusion in the negligence litigation with most others declining press comment.
Twitter: @eriqgardner
Must Reads
Bar-Hopping With Kyle Chandler: 'Friday Night Lights' Star on His "Dark, Evil" Period, Comedy Dreams and Return to TV
How Leonard Nimoy Was Convinced to Join the First 'Star Trek' Movie
'Deer Hunter' Director's First Interview in 13 Years: Michael Cimino Touts Clint Eastwood for President, Blasts Rumors He's Becoming a Woman
The Day Chris Kyle Died: Text Messages and Terror in the 'American Sniper's' Final Hours With Killer Eddie Ray Routh
'Parks and Rec' Writer Harris Wittels' Death at 30: Humblebrags and the Specter of Heroin
Follow Esq.
- Mo'Nique Says She Was Supposed to Play Empire's Cookie First
- Felicity Huffman on American Crime and Playing an Unlikable Character
- Read an Annotated Excerpt From Adam Rapp's Novel Know Your Beholder
- Vince Vaughn Got Photoshopped Into Cheesy Stock Photos Because This Is How You Promote a Movie in 2015


text:
Emma Stone
6 November 1988, Scottsdale, Arizona, USA26 years
Emma Stone appered in the comedy film The Rocker in 2008 and will also appear in the film The Ghosts of girlfriends past in 2009. Since 2007 Emma lives on her on in Los Angeles and has stated that she is skilled in HTML and has web design as a hobby. Although she's known as a redhead., her natural hair color is blonde. Emma Stone convinced her parents to let her move to Hollywood when she was fourteen by making a Powerpoint presentation aptly titled, Project Hollywood.
RSS feed with Emma Stone
Complete List! Check Out The Best Dressed Women Of 2013!
Monday 9:05 PM, 06/05/2013
Every year, Glamour UK name the best dressed female celebrities of the year, and this year we were really surprised by the list. It consists of women of all ages from the business, but they have one thing in common: they know how to dress to impress! Click on the pics and check out the top 10 best dressed women of 2013!
Top Lists
Latest news
Emma Stone's Twitter Account Is Hacked!
Friday 3:27 PM, 22/03/2013
If you follow Emma Stone on Twitter, you might've noticed she's tweeted some very mysterious things, implying that her relationship with And...READ MORE ▶
Hot News
Our Top 10 Favorite Oscars Dresses Of All Time!
Friday 10:56 AM, 22/02/2013
As you guys might already know, the annual Academy Awards (Oscars) takes place this weekend, and it really is the perfect moment to check out what all the superstars are wearing at the red carpet. Some dresses are still in our minds, years later, so we figured we'd list our top 10 favorite Oscars dresses of all time! Click on the pics to see them all!
Valentine's Day Special!
Happy Valentine's Day! Top 6 Romantic Movies!
Thursday 7:43 PM, 14/02/2013
It's the 14th of February 2013, which can only mean one thing: it's Valentine's Day! It's the day of Love, and we wanted to give you guys a list with our top 6 best movie tips to celebrate the day! By clicking on the pics you'll get the inspo for tonight's cozy or romantic evening. Happy Valentine's Day, everyone!
Top Lists
Valentine's Day Special!
Celebrity Couples Showing Their True Feelings In Public!
Thursday 12:26 PM, 14/02/2013
We love really cute celebrity couples and especially those who show their true feelings for each other, no matter who's watching! We've listed six super cute Hollywood couples that we hope will last forever and ever! Click on the pics and see our favorite couples!
Celebrity couples