Danial David (Editor)

Distributed System

Updated on
Edit
Like
Comment
Share on FacebookTweet on TwitterShare on LinkedInShare on Reddit
Distributed computing (DC)

http://static.trustedreviews.com/94/3e2601/1e8f/8476-distributedcomputing.jpg
Fundamentals
Distributed computing
Distributed computing is a field of computer science that studies distributed systems. A distributed system is a software system in which components located on networked computers communicate and coordinate their actions by passing messages. The components interact with each other in order to achieve a common goal. Three significant characteristics of distributed systems are: concurrency of components, lack of a global clock, and independent failure of components. Examples of distributed systems vary from SOA-based systems to massively multiplayer online games to peer-to-peer applications.

A computer program that runs in a distributed system is called a distributed program, and distributed programming is the process of writing such programs. There are many alternatives for the message passing mechanism, including RPC-like connectors and message queues. An important goal and challenge of distributed systems is location transparency.

Distributed computing also refers to the use of distributed systems to solve computational problems. In distributed computing, a problem is divided into many tasks, each of which is solved by one or more computers, which communicate with each other by message passing.

The word distributed in terms such as "distributed system", "distributed programming", and "distributed algorithm" originally referred to computer networks where individual computers were physically distributed within some geographical area. The terms are nowadays used in a much wider sense, even referring to autonomous processes that run on the same physical computer and interact with each other by message passing. While there is no single definition of a distributed system, the following defining properties are commonly used:

1. There are several autonomous computational entities, each of which has its own local memory.

2. The entities communicate with each other by message passing.

the computational entities are called computers or nodes.

A distributed system may have a common goal, such as solving a large computational problem. Alternatively, each computer may have its own user with individual needs, and the purpose of the distributed system is to coordinate the use of shared resources or provide communication services to the users.

Other typical properties of distributed systems include the following:

1. The system has to tolerate failures in individual computers.

2. The structure of the system (network topology, network latency, number of computers) is not known in advance, the system may consist of different kinds of computers and network links, and the system may change during the execution of a distributed program.

3. Each computer has only a limited, incomplete view of the system. Each computer may know only one part of the input.

Message Passing


Message passing is a concept from computer science that is used extensively in the design and implementation of modern software applications; it is key to some models of concurrency and object-oriented programming. Message passing is a way of invoking behavior through some intermediary service or infrastructure. Rather than directly invoke a process, subroutine, or function by name as in conventional programming, message passing sends a message to a process (which may be an actor or object) and relies on the process and the supporting infrastructure to select and invoke the actual code to run.

Message passing is used ubiquitously in modern computer software. It is used as a way for the objects that make up a program to work with each other and as a way for objects and systems running on different computers (e.g., the Internet) to interact. Message passing may be implemented by various mechanisms, including channels.

Message passing is a technique for invoking behavior (i.e., running a program) on a computer. In contrast to the traditional technique of calling a program by name, message passing uses an object model to distinguish the general function from the specific implementations. The invoking program sends a message and relies on the object to select and execute the appropriate code. The justifications for using an intermediate layer essentially falls into two categories: encapsulation and distribution.

Encapsulation is the idea that software objects should be able to invoke services on other objects without knowing or caring about how those services are implemented. Encapsulation can reduce the amount of coding logic and make systems more maintainable. E.g., rather than having IF-THEN statements that determine which subroutine or function to call a developer can just send a message to the object and the object will select the appropriate code based on its type.

One of the first examples of how this can be used was in the domain of computer graphics. There are all sorts of various complications in manipulating graphic objects. For example, simply using the right formula to compute the area of an enclosed shape will vary depending on if the shape is a triangle, rectangle, elipse, or circle. In traditional computer programming this would result in long IF-THEN statements testing what sort of object the shape was and calling the appropriate code. The object-oriented way to handle this is to define a class called Shape with subclasses such as Rectangle and Ellipse (which in turn have subclasses Square and Circle) and then to simply send a message to any Shape asking it to compute its area. Each Shape object will then invoke the appropriate code with the formula appropriate for that kind of object.

Distributed message passing provides developers with a layer of the architecture that provides common services to build systems made up of sub-systems that run on disparate computers in different locations and at different times. When sending a distributed object a message the messaging layer can take care of issues such as:

1. Finding the appropriate object, including objects running on different computers, using different operating systems and programming languages, at different locations from where the message originated.
2. Saving the message on a queue if the appropriate object to handle the message is not currently running and then invoking the message when the object is available. Also, storing the result if needed until the sending object is ready to receive it.
3. Controlling various transactional requirements for distributed transactions, e.g. ensuring ACID properties on data.

Synchronous versus asynchronous message passing

One of the most important distinctions among message passing systems is whether they use synchronous or asynchronous message passing. Synchronous message passing occurs between objects that are running at the same time. With asynchronous message passing it is possible for the receiving object to be busy or not running when the requesting object sends the message.

Synchronous message passing is what typical object-oriented programming languages such as Java and Smalltalk use. Asynchronous message passing requires additional capabilities for storing and retransmitting data for systems that may not run concurrently.

The advantage to synchronous message passing is that it is conceptually less complex. Synchronous message passing is analogous to a function call in which the message sender is the function caller and the message receiver is the called function. Function calling is easy and familiar. Just as the function caller stops until the called function completes, the sending process stops until the receiving process completes. This alone makes synchronous message unworkable for some applications. For example, if synchronous message passing would be used exclusively, large, distributed systems generally would not perform well enough to be usable. Such large, distributed systems may need to continue to operate while some of their subsystems are down; subsystems may need to go offline for some kind of maintenance, or have times when subsystems are not open to receiving input from other systems.

Imagine a busy business office having 100 desktop computers that send emails to each other using synchronous message passing exclusively. Because the office system does not use asynchronous message passing, one worker turning off their computer can cause the other 99 computers to freeze until the worker turns their computer back on to process a single email.

Asynchronous message passing is generally implemented so that all the complexities that naturally occur when trying to synchronize systems and data are handled by an intermediary level of software. Commercial vendors who develop software products to support these intermediate levels usually call their software "middleware". One of the most common types of middleware to support asynchronous messaging is called Message Oriented Middleware (MOM)

http://rtcmagazine.com/archive_images/rtc0509sd_eet2.gif
With asynchronous message passing, the sending system does not wait for a response. Continuing the function call analogy, asynchronous message passing would be a function call that returns immediately, without waiting for the called function to execute. Such an asynchronous function call would merely deliver the arguments, if any, to the called function, and tell the called function to execute, and then return to continue its own execution. Asynchronous message passing simply sends the message to the message bus. The bus stores the message until the receiving process requests messages sent to it. When the receiving process arrives at the result, it sends the result to the message bus. And the message bus holds the message until the original process (or some designated next process) picks up its messages from the message bus.[3]

Synchronous communication can be built on top of asynchronous communication by using a Synchronizer. For example, the ?-Synchronizer works by ensuring that the sender always waits for an acknowledgement message from the receiver. The sender only sends the next message after the acknowledgement has been received.

The buffer required in asynchronous communication can cause problems when it is full. A decision has to be made whether to block the sender or whether to discard future messages. If the sender is blocked, it may lead to an unexpected deadlock. If messages are dropped, then communication is no longer reliable. These are all examples of the kinds of problems that middleware vendors try to address.

Message passing in computer science is a form of communication used in distributed computing, in which processes or objects can send and receive messages (comprising zero or more bytes, complex data structures, or even segments of code) to other processes.

Message passing systems have been called "shared nothing" systems because the message passing abstraction hides underlying state changes that may be used in the implementation of sending messages.

Desirable Features of a Good Message-Passing System

1) Simplicity
a)  A message passing system should be simple and easy to use.
b) It should be possible to communicate with old and new applications, with different modules without the need to worry about the system and network aspects.

2) Uniform Semantics
Semantics of remote communication should be as close as possible to those of local communications.
 
3) Efficiency
less number of message exchanges, as far as practicable, during the communication process.

4) Correctness
a)  Atomicity – message sent to a group of receivers will be delivered to either all of them or none of them.
b) Ordered delivery – messages arrive to all receivers in an order acceptable to the application
c)  Survivability - messages will be correctly delivered despite partial failures of processes, machines, or communication links.

5) Reliability

6) Flexibility

7) Security

8) Portability

Issues in IPC by Message Passing

 A message is a block of information formatted by a sending process in such a manner that it is meaningful to the receiving process.

It consists of a fixed-length header and a variable-size collection of typed data objects. The header usually consists of the following elements: Address, Sequence Number, Structural Information.

Remote procedure call
In computer science, a remote procedure call (RPC) is an inter-process communication that allows a computer program to cause a subroutine or procedure to execute in another address space (commonly on another computer on a shared network) without the programmer explicitly coding the details for this remote interaction.That is, the programmer writes essentially the same code whether the subroutine is local to the executing program, or remote. When the software in question uses object-oriented principles, RPC is called remote invocation or remote method invocation.
http://jan.newmarch.name/go/rpc/rpc_stub.png

Sequence of events during an RPC

1. The client calls the client stub. The call is a local procedure call, with parameters pushed on to the stack in the normal way.
2. The client stub packs the parameters into a message and makes a system call to send the message. Packing the parameters is called marshalling.
3. The clients local operating system sends the message from the client machine to the server machine.
4. The local operating system on the server machine passes the incoming packets to the server stub.
5. The server stub unpacks the parameters from the message. Unpacking the parameters is called unmarshalling.
6. Finally, the server stub calls the server procedure. The reply traces the same steps in the reverse direction.

RPC Model
The RPC mechanism maps the local procedure call paradigm onto an environment where the calling procedure and the called procedure are distributed between different execution contexts that usually, but not necessarily, reside on physically separate computers that are linked by communications networks.

A procedure is defined as a closed sequence of instructions that is entered from, and returns control to, an external source. Data values may be passed in both directions along with the flow of control. A procedure call is the invocation of a procedure. A local procedure call and an RPC behave similarly; however, there are semantic differences due to several properties of RPCs:

1. Server/client relationship (binding)
While a local procedure call depends on a static relationship between the calling and the called procedure, the RPC paradigm requires a more dynamic behaviour. As with a local procedure call, the RPC establishes this relationship through binding between the calling procedure (client) and the called procedure (server). However, in the RPC case a binding usually depends on a communications link between the client and server RPC run-time systems. A client establishes a binding over a specific protocol sequence to a specific host system and endpoint.

2. No assumption of shared memory
Unlike a local procedure call, which commonly uses the call-by-reference passing mechanism for input/output parameters, RPCs with input/output parameters have copy-in, copy-out semantics due to the differing address spaces of calling and called procedures.

3. Independent failure
Beyond execution errors that arise from the procedure call itself, an RPC introduces additional failure cases due to execution on physically separate machines. Remoteness introduces issues such as remote system crash, communications links, naming and binding issues, security problems, and protocol incompatibilities.

4. Security
 Executing procedure calls across physical machine boundaries has additional security implications. Client and server must establish a security context based on the underlying security protocols, and they require additional attributes for authorising access.

Client/Server Execution Model
The RPC model makes a functional distinction between clients and servers. A client requests a service, and a server provides the service by making resources available to the remote client.

RPC Interface and RPC Object
Two entities partially determine the relationship between a client and a server instance: RPC interfaces and RPC objects. Both interfaces and objects are identified by UUIDs. (See Universal Unique Identifier for a UUID specification.)

Types of remote procedure call

These are the five types of remote procedure call.

Synchronous
This is the normal method of operation. The client makes a call and does not continue until the server returns the reply.

Nonblocking
The client makes a call and continues with its own processing. The server does not reply.

Batching
This is a facility for sending several client nonblocking calls in one batch.

Broadcast RPC
RPC clients have a broadcast facility, that is, they can send messages to many servers and then receive all the consequent replies.

Callback RPC
The client makes a nonblocking client/server call, and the server signals completion by calling a procedure associated with the client.
   
 CICS ONC RPC: CICS® ONC RPC cannot support callback RPC, because callback requires that both ends contain both client and server procedures.

https://www.cs.rutgers.edu/~pxk/417/notes/images/rpc-flow.png
Lightweight RPC
Lightweight Remote Procedure Call (LRPC) is a communication facility designed and optimized for communication between protection domains on the same machine. In contemporary small-kernel operating systems, existing RPC systems incur an unnecessarily high cost when used for the type of communication that predominates?between protection domains on the same machine. This cost leads system designers to coalesce weakly related subsystems into the same protection domain, trading safety for performance. By reducing the overhead of same-machine communication, LRPC encourages both safety and performance. LRPC combines the control transfer and communication model of capability systems with the programming semantics and large-grained protection model of RPC. LRPC achieves a factor-of-three performance improvement over more traditional approaches based on independent threads exchanging messages, reducing the cost of same-machine communication to nearly the lower bound imposed by conventional hardware. LRPC has been integrated into the Taos operating system of the DEC SRC Firefly multiprocessor workstation.

Distributed Shared Memory
In computer architecture, distributed shared memory (DSM) is a form of memory architecture where the (physically separate) memories can be addressed as one (logically shared) address space. Here, the term shared does not mean that there is a single centralized memory but shared essentially means that the address space is shared (same physical address on two processors refers to the same location in memory).[1] Distributed Global Address Space (DGAS), is a similar term for a wide class of software and hardware implementations, in which each node of a cluster has access to shared memory in addition to each nodes non-shared private memory.

Software DSM systems can be implemented in an operating system (OS), or as a programming library and can be thought of as extensions of the underlying virtual memory architecture. When implemented in the OS, such systems are transparent to the developer; which means that the underlying distributed memory is completely hidden from the users. In contrast, Software DSM systems implemented at the library or language level are not transparent and developers usually have to program differently. However, these systems offer a more portable approach to DSM system implementation.

Software DSM systems also have the flexibility to organize the shared memory region in different ways. The page based approach organizes shared memory into pages of fixed size. In contrast, the object based approach organizes the shared memory region as an abstract space for storing shareable objects of variable sizes. Another commonly seen implementation uses a tuple space, in which the unit of sharing is a tuple.

Shared memory architecture may involve separating memory into shared parts distributed amongst nodes and main memory; or distributing all memory between nodes. A coherence protocol, chosen in accordance with a consistency model, maintains memory coherence.

Examples of such systems include:
Kerrighed
OpenSSI
MOSIX
TreadMarks
DIPC

Synchronization
Synchronization is the coordination of events to operate a system in unison. The familiar conductor of an orchestra serves to keep the orchestra in time. Systems operating with all their parts in synchrony are said to be synchronous or in sync.

Today, synchronization can occur on a global basis through the GPS-enabled timekeeping systems.

clock synchronization
http://www.ni.com/cms/images/devzone/tut/c/c8bd1a32550.gif
Clock synchronization is a problem from computer science and engineering which deals with the idea that internal clocks of several computers may differ. Even when initially set accurately, real clocks will differ after some amount of time due to clock drift, caused by clocks counting time at slightly different rates. There are several problems that occur as a repercussion of clock rate differences and several solutions, some being more appropriate than others in certain contexts.

In serial communication, some people  use the term "clock synchronization" merely to discuss getting one metronome-like clock signal to pulse at the same frequency as another one – frequency synchronization (plesiochronous or isochronous operation), as opposed to full phase synchronization (synchronous operation). Such "clock synchronization" is used in synchronization in telecommunications and automatic baud rate detection.

mutual exclusion     
In computer science, mutual exclusion refers to the requirement of ensuring that no two concurrent processes[a] are in their critical section at the same time; it is a basic requirement in concurrency control, to prevent race conditions. Here, a critical section refers to a period when the process accesses a shared resource, such as shared memory. The requirement of mutual exclusion was first identified and solved by Edsger W. Dijkstra in his seminal 1965 paper titled Solution of a problem in concurrent programming control,  and is credited as the first topic in the study of concurrent algorithms.

A simple example of why mutual exclusion is important in practice can be visualized using a singly linked list (See Figure 1). In such a linked list, the removal of a node is done by changing the “next” pointer of the preceding node to point to the subsequent node (e.g., if node i is being removed then the “next” pointer of node i?1 will be changed to point to node i+1). In an execution where such a linked list is being shared between multiple processes, two processes may attempt to remove two different nodes simultaneously, resulting in the following problem: let nodes i and i+1 be the nodes to be removed; furthermore, let neither of them be the head nor the tail; the next pointer of node i?1 will be changed to point to node i+1 and the next pointer of node i will be changed to point to node i+2. Although both removal operations complete successfully, node i+1 remains in the list since i?1 was made to point to i+1, skipping node i (which was the node that reflected the removal of i+1 by having its next pointer set to i+2). This can be seen in Figure 1. This problem (normally called a race condition) can be avoided by using the requirement of mutual exclusion to ensure that simultaneous updates to the same part of the list cannot occur.

Deadlock
A deadlock is a situation in which two or more competing actions are each waiting for the other to finish, and thus neither ever does.

In a transactional database, a deadlock happens when two processes each within its own transaction updates two rows of information but in the opposite order. For example, process A updates row 1 then row 2 in the exact timeframe process B updates row 2 then row 1. Process A cant finish updating row 2 until process B is finished, but it cannot finish updating row 1 until process A finishes. No matter how much time is allowed to pass, this situation will never resolve itself and because of this database management systems will typically kill the transaction of the process that has done the least amount of work.


In an operating system, a deadlock is a situation which occurs when a process or thread enters a waiting state because a resource requested is being held by another waiting process, which in turn is waiting for another resource. If a process is unable to change its state indefinitely because the resources requested by it are being used by another waiting process, then the system is said to be in a deadlock.

Deadlock is a common problem in multiprocessing systems, parallel computing and distributed systems, where software and hardware locks are used to handle shared resources and implement process synchronization.

In telecommunication systems, deadlocks occur mainly due to lost or corrupt signals instead of resource contention.

Distributed deadlock
Distributed deadlocks can occur in distributed systems when distributed transactions or concurrency control is being used. Distributed deadlocks can be detected either by constructing a global wait-for graph from local wait-for graphs at a deadlock detector or by a distributed algorithm like edge chasing.

Phantom deadlocks are deadlocks that are falsely detected in a distributed system due to system internal delays but dont actually exist. For example, if a process releases a resource R1 and issues a request for R2, and the first message is lost or delayed, a coordinator (detector of deadlocks) could falsely conclude a deadlock (if the request for R2 while having R1 would cause a deadlock).

Resource and Process Management

Resource Management
Resources required for manufacturing include materials, equipment, and personnel. Syncade suite provides an operations management system that effectively manages these resources, allowing you to do more with less. Scheduling and tracking equipment usage can increase capacity. Optimizing material usage and eliminating the use of outdated material can reduce waste and rework and assure production meets its targeted specifications and yield. Improving operator productivity by guiding manual processes and enabling access to support documents can increase plant performance. Ensuring personnel have the training and qualifications needed to effectively perform their jobs can reduce costly mistakes and assure regulatory compliance.

Load balancing

http://manikumarreddy.files.wordpress.com/2013/12/load-balancing1-fig1.gif
Load balancing is a computer networking method for distributing workloads across multiple computing resources, such as computers, a computer cluster, network links, central processing units or disk drives. Load balancing aims to optimize resource use, maximize throughput, minimize response time, and avoid overload of any one of the resources. Using multiple components with load balancing instead of a single component may increase reliability through redundancy. Load balancing is usually provided

Introduction to process management
processes

A process is the activity on a system caused by a running program.

UNIX is a multitasking system, which means it has facilities for controlling and tracking multiple jobs or processes at the same time, and ensuring they get their appropriate share of system resources such as CPU time. It is also a multi-user operating system which means that it can simultaneously manage files and processes belonging to more than one user on the same system. Security features in the operating system prevent these processes from interfering with each other. The kernel gives the impression of keeping multiple processes active simultaneously by switching between them more rapidly than the eye can see.

Each process has certain information associated with it including:

1. The UID (numeric user identity)
2. The GID (numeric group identity)
3. A process ID number used to identify the process
4. A parent process ID
5. The execution status, e.g. active, runnable, waiting for input etc.
6. Environment variables and values.
7. The current directory.
8. Where the process currently resides in memory
9. The relative process priority see nice(1)
10. where it gets standard input from
11. where it sends standard output to
12.any other files currently open

process migration

check out following link for more info about process migration


In computing, process migration is a specialized form of process management whereby processes are moved from one computing environment to another. The most common application of process migration is in computer clusters where processes are moved from machine to machine. Process migration is implemented in, among others, OpenMosix. It was pioneered by the Sprite OS from the University of California, Berkeley.

Process migration in computing comes in two flavors:
Non-preemptive process migration
    Process migration that takes place before execution of the process starts (i.e. migration whereby a process need not be preempted). This type of process migration is relatively cheap, since relatively little administrative overhead is involved.

Preemptive process migration
    Process migration whereby a process is preempted, migrated and continues processing in a different execution environment. This type of process migration is relatively expensive, since it involves recording, migration and recreation of the process state as well as the reconstructing of any inter-process communication channels to which the migrating process is connected.

An alternate definition of process migration is used in integrated circuit design and engineering. Process migration or layout migration in this context is a design flow to change and shrink an existing IC layout to a new process technology node. The implementation of a process migration could be done manually by redrawing the layout feature by feature or by automatic EDA/CAD tools. In load sharing systems, a process is migrated from one node to another using a mechanism called process migration.

Distributed File Systems
A method of storing and accessing files based in a client/server architecture. In a distributed file system, one or more central servers store files that can be accessed, with proper authorization rights, by any number of remote clients in the network. Much like an operating system organizes files in a hierarchical file management system, the distributed system uses a uniform naming convention and a mapping scheme to keep track of where files are located. When the client device retrieves a file from the server, the file appears as a normal file on the client machine, and the user is able to work with the file in the same ways as if it were stored locally on the workstation. When the user finishes working with the file, it is returned over the network to the server, which stores the now-altered file for retrieval at a later time.

Distributed file systems can be advantageous because they make it easier to distribute documents to multiple clients and they provide a centralized storage system so that client machines are not using their resources to store files.

NFS from Sun Microsystems and Dfs from Microsoft are examples of distributed file systems

File sharing Semantics

http://cs.gmu.edu/~menasce/osbook/distfs/img041.GIF

Naming
The naming faciliy of a distributed operating system enables users and programs to assign character string names to objects and subsequently use these names to refer to those objects.

The naming and locating facilities jointly form a naming system that provides the users with an abstraction of an object that hides the details of how and where an object is actually located in the network.

The naming system plays a very important role in achieving the goal of
1. location transparency,
2. facilitating transparent migration and replication
of objects,
3. object sharing.

check out this site
http://mazsola.iit.uni-miskolc.hu/DATA/research/tempus/discom/doc/os/obsolete/naming_3.pdf

DESIRABLE FEATURES OF A GOOD NAMING SYSTEM

1. Location transparency. Location transparency means that the name of an object should not reveal any hint as to the physical location of the object. That is, an objects name should be independent of the physical connectivity or topology of the system, or the current location of the object.

2. Location independency. For performance, reliability, availability, and security reasons, distributed systems provide the facility of object migration  that allows th movement and relocation of objects dynamically among the various nodes of a system. Location independency means that the name of an object need not be changed when the objects location changes.

Furthermore, a user should be able to access an object by its same name irrespective of the node from where he or she accesses it ( user migration ).

Name Caches and Naming and Security
If multiple users will be using an application that is sharing classes or multiple applications are sharing the same cache, knowing how to name caches appropriately is important. The ultimate goal is to have the smallest number of caches possible, while maintaining secure access to the class data and allowing as many applications and users as possible to share the same classes.
http://cs.gmu.edu/~menasce/osbook/distfs/img051.GIF
To use a cache for a specific application, write the cache into the application installation directory using the -Xshareclasses:cachedir=<dir> option. This helps prevent users of other applications from accidentally using the same cache, and automatically removes the cache if the application is uninstalled.

If the same user will always be using the same application, either use the default cache name (which includes the user name) or specify a cache name specific to the application. The user name can be incorporated into a cache name using the %u modifier, which causes each user running the application to get a separate cache.

On Linux®, AIX®, z/OS®, and i5/OS® platforms, if multiple users in the same operating system group are running the same application, use the groupAccess suboption, which creates the cache allowing all users in the same primary group to share the same cache. If multiple operating system groups are running the same application, the %g modifier can be added to the cache name, causing each group running the application to get a separate cache.

Multiple applications or different JVM installations can share the same cache provided that the JVM installations are of the same service release level. It is possible for different JVM service releases to share the same cache, but it is not advised. The JVM will attempt to destroy and re-create a cache created by a different service release. See Compatibility between service releases for more information.

Small applications that load small numbers of application classes should all try to share the same cache, because they will still be able to share bootstrap classes. For large applications that contain completely different classes, it might be more sensible for them to have a class cache each, because there will be few common classes and it is then easier to selectively clean up caches that arent being used.

On Linux, AIX, z/OS, and i5/OS, /tmp is used as the default directory, which is shared by all users.





Similar Topics