System and method for replicating, integrating and synchronizing distributed information

• US 20050086384A1
• Filed: 09/03/2004
• Published: 04/21/2005
• Est. Priority Date: 09/04/2003
• Status: Abandoned Application

• Current Assignee: R-Objects Incorporated
• Original Assignee: R-Objects Incorporated
• Inventors: Ernst, Johannes
• Application Number:
  US10/934,206
• Publication Number:
  US 20050086384A1
• Time in Patent Office: Days
• US Class Current: 709/248
• CPC Class Codes: G06F 16/00 Information retrieval; Database structures therefor; File system structures therefor
• CPC Class Codes: G06F 16/178 Techniques for file synchronisation in file systems
• CPC Class Codes: G06F 16/184 implemented as replicated file system
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor
• CPC Class Codes: G06F 16/275 Synchronous replication
• CPC Class Codes: H04L 12/00 Data switching networks interconnection of, or transfer of information or other signals between, memories, input/output devices or central processing units G06F13/00
• CPC Class Codes: H04L 12/4625 Single bridge functionality, e.g. connection of two networks over a single bridge
• CPC Class Codes: H04L 67/1095 Replication or mirroring of data, e.g. scheduling or transport for data synchronisation between network nodes

Abstract

An extensible protocol to replicate, integrate and synchronize distributed information is described which may be implemented in a computer system. A system and method for replicating, integrating and synchronizing distributed information is also described.

Claims

1. A distributed system for sharing a plurality of related pieces of information, comprising two or more heterogeneous nodes capable of being connected to each other wherein the nodes exchange a plurality of messages according to a common communication protocol that runs concurrently on a plurality of reliable or non-reliable communication transports between the nodes, wherein the shared pieces of information may be updated frequently by one or more of the nodes and wherein the shared information is kept coherent across the nodes.

2. The distributed system of claim 1, wherein the shared information further comprises one or more entity objects and one or more relationship objects, each entity object and each relationship object being identified using a unique identifier.

3. The distributed system of claim 2, wherein each entity object and each relationship object further comprises one or more properties, each of which carries atomic information.

4. The distributed system of claim 3 further comprising an information model, agreed to by the nodes of the distributed system, that governs the structure of the shared information for the purpose of sharing it among the nodes, and for when nodes communicate with each other about the shared information, wherein the entity objects, relationship objects and their properties are instances of the information model.

5. The distributed system of claim 4, wherein the information model further comprises a model that defines the structure and relationships of configuration management and version control information.

6. The distributed system of claim 4, wherein the information model further comprises a model that defines the structure and relationships of access control information for pieces of shared information, and wherein the pieces of shared information are subject to the access control rules represented by the pieces of shared information that are instances of the access control information concepts in the information model.

7. The distributed system of claim 4, wherein the information model is fixed prior to commencing operation of the distributed system.

8. The distributed system of claim 4, wherein a core part of the information model is fixed prior to commencing operation of a first instance of the distributed system and wherein the nodes may dynamically discover and support other parts of the information model during operation by gaining knowledge of the other parts of the information model from one of the other nodes or an information model distribution facility.

9. The distributed system of claim 8, further comprising a second instance of the distributed system, running concurrently with the first instance of the distributed system wherein each node in the first instance of the distributed system is associated with exactly one node of the second instance of the distributed system, the second instance of the distributed system being governed by a “

meta”

information model that the nodes of the second instance of the distributed system agree on, and sharing the information model of the first instance as the second instance'"'"'s shared information, and the first instance of the distributed system using the information model shared as the shared information by the second instance as its information model.

10. The distributed system of claim 1, wherein each of the nodes of the distributed system holds a replica of each piece of shared information.

11. The distributed system of claim 1, wherein each node of a first portion of the nodes of the distributed system holds a replica of all of the pieces of the shared information, and each node of a second portion of the nodes of the distributed system holds a replica of only some of the pieces of the shared information, with different nodes in the second portion holding replicas of different pieces of the shared information.

12. The distributed system of claim 1, wherein none of the nodes of the distributed system holds a replica of all of the pieces of the shared information.

13. The distributed system of claim 1, wherein none of the nodes of the distributed system holds a replica of all of the pieces of shared information for security reasons, and wherein some of the replicas of some of the pieces of the shared information at some nodes are of an incomplete type.

14. The distributed system of claim 1, wherein each piece of shared information has a home node associated with the piece of shared information, the home node being the same for each replica of the same piece of shared information, and wherein the replica of the piece of shared information held by the home node is called a home replica.

15. The distributed system of claim 14, wherein each piece of shared information further comprises one or more replicas of the piece of shared information wherein each replica that is not the home replica for the piece of the shared information is subject to a lease that is negotiated between a granting node, being one of the home node and another node having a replica of the piece of shared information, and the node holding the replica wherein the lease has a duration up to an expiration time during which the replica is coherent with the replica of the piece of shared information at the granting node.

16. The distributed system of claim 15, wherein one or more leases for replicas, between the same granting node and the same node receiving the leases, are grouped together in a lease group wherein each lease for each replica has the same expiration time.

17. The distributed system of claim 16, wherein the lease group is split into a first and second new lease group wherein the replicas from the lease group are split into the first and second lease groups.

18. The distributed system of claim 15, wherein a node holding a replica requests an extension of the lease for the replica prior to the expiration time from the granting node, wherein the granting node one of grants and denies the request for the lease extension.

19. The distributed system of claim 15, wherein each granting node notifies each node, to which there is a lease of a replicathat has not expired, of changes to the particular piece of shared information and the node from which it has been granted a lease that has not expired yet for the particular replica.

20. The distributed system of claim 15, wherein exactly one node, holding a replica of a particular piece of the shared information, holds a lock for the particular piece of shared information, and wherein changes to the particular piece of shared information are only made to the replica held by the node currently holding the lock for the said particular piece of the shared information.

21. The distributed system of claim 20, wherein the granting node has granted a lease for a replica for piece of shared information to a second node wherein the lease has expired without having been successfully renewed, and the second node possessing the lock for the piece of shared information at the time of lease expiration, the granting node unilaterally retrieving the lock for its own replica of the piece of the shared information once the lease has expired.

22. The distributed system of claim 14, wherein the designation as the home node associated with a piece of shared information may be moved between nodes during the operation of the distributed system.

23. The distributed system of claim 3, wherein each granting node notifies each node, to which there is a lease of a particular piece of information that has not expired, of changes to the particular piece of shared information and the node from which it has been granted a lease that has not expired yet for the particular piece of the shared information and wherein the notification of changes to a first shared entity object comprises one or more of notification of a change of one or more of the properties of the first shared entity object and the new values for the properties, notification of the creation of a relationship object relating to the first shared entity object, notification of deletion of a relationship object relating to the first shared entity object, notification of change of a relationship object relating to the first shared entity object, and notification of deletion of the first shared entity object;

and where notification of changes to a second shared relationship object comprises notification of change of one or more of the properties of the second shared relationship object and the new values for the properties, notification of deletion of an entity object that is related to the second shared relationship object, and notification of deletion of the second shared relationship object.

24. The distributed system of claim 15, wherein a node holds a zombie replica which is a replica whose lease has expired and wherein the node requests, from another node, a revival of the zombie replica and the other node grants or denies the zombie revival request.

25. The distributed system of claim 1, wherein each node is identified by one unique node identifier, resolvable by a network transport and routable from the other nodes in the distributed system, for each network transport that may be employed by the node.

26. The distributed system of claim 1, wherein each message is expressed in a XML format.

27. The distributed system of claim 3, wherein the property values are contained within the message.

28. The distributed system of claim 3, wherein property values form a main part of the message and non-property information is quoted within the message.

29. The distributed system of claim 1, wherein a message comprises a plurality of requests and a plurality of responses.

30. The distributed system of claim 1, wherein one or more nodes become unavailable after the operation of the distributed system has commenced, and wherein one or more nodes join the distributed system after the operation of the distributed system has commenced.

31. The distributed system of claim 1, wherein one of the nodes is a test node to test the operation of one or more nodes and of the distributed system.

32. The distributed system of claim 15, wherein each node of the distributed system further comprises:

an information storage unit that stores the replica of one or more pieces of shared information, a first portion of the replicas having non-home replicas subject to a lease from a granting node and a second portion of the replicas being the home replicas;

a piece of lock information for each replica indicating which node of the distributed system has a right to update the piece of shared information; and

a piece of lease information for each non-home replica indicating the duration of the lease for the non-home replica and the granting node for the lease;

a transaction serializer that guards the information storage unit against unmanaged concurrent access;

a lease manager further comprising a unit that attempts to renew the lease, approaching the expiration time, of the replicas needed by the node, a unit that destroys, when the lease is not renewed, replicas subject to the lease after the expiration time and a unit, when the node is the granting node, that grants or denies requests for renewed leases from other nodes; and

one or more protocol managers wherein each protocol manager is responsible for a particular communication transport, each protocol manager receives incoming messages and converts the incoming message into an internal protocol, and sends outgoing messages wherein the outgoing message is converted from the internal protocol to the particular communication transport protocol;

one or more proxy units connected to the information storage unit and to one or more of the protocol managers, each proxy unit controlling access, between the node and a second node, to the plurality of replicas stored in the information storage unit, each proxy unit receiving incoming messages using the internal protocol from one or more protocol managers, sending outgoing messages to the protocol manager, detecting that incoming messages were lost during transport, and creating, sending, receiving and managing messages that request from other nodes to resend messages they sent, and that responds to incoming requests from other nodes to resend messages;

one or more priority queues, one for each proxy unit, that store incoming messages in the internal protocol according to the sequence in which they were created by a sending node; and

for each proxy unit, a set of messages that were sent by that proxy to another node but whose receipt has not been acknowledged yet by the other node and a set of messages that were received from another node, but whose receipt the node has not acknowledged yet to the other node.

33. The system of claim 32, wherein the node holds a zombie replica which is a replica whose lease has expired and wherein the lease manager further comprises a unit that initiates and responds to zombie revival requests.

34. The system of claim 32, wherein the proxy unit further comprises a unit to send a given message in multiple copies to a destination node through multiple communication transports, receive such messages from the destination node through multiple communication transports, and discard all but one copy of the received messages.

35. The system of claim 32, wherein the proxy unit further comprises a message confirmation unit that confirms receipt of each message originating from the proxy unit to another node and confirms receipt of each incoming message from another node to perform a message handshaking protocol.

36. The system of claim 32, further comprising a lease manager that manages leases to other nodes and leases obtained from other nodes by grouping replicas into a lease group with the same expiration times and granting node.

37. The system of claim 32, wherein each proxy unit, in response to a replication request from a requesting node, grants a lease to the requesting node for all replicas available at the node.

38. The system of claim 32, wherein each proxy unit, in response to a replication request from a requesting node, grants a lease to a portion of the replicas available at the node to the requesting node.

39. The system of claim 38, wherein the proxy unit determines a portion of the replicas available at the node to which it grants a lease to the requesting node by partitioning the replicas of the entity objects into one or more newly-shared “

complete”

replicas of entity objects, newly-shared “

incomplete”

replicas of entity objects, not-shared replicas of entity objects, already-shared but newly-referenced replicas of entity objects, newly-shared replicas of relationship objects, already-shared but newly-referenced replicas of relationship objects, and not-shared replicas of relationship objects.

40. The system of claim 39, wherein the proxy unit further comprises means for converting an “

incomplete”

replica of entity objects into a “

complete”

replica by determining a set of relationships related to the entity object, and then obtaining replicas of such related relationships, from other replicas of the same entity object at other nodes from which the node has a lease.

41. The system of claim 39 that partitions entity objects and relationship objects according to a scope parameter provided by the requesting node.

42. The system of claim 32, wherein the proxy unit accumulates outgoing change notifications for a shared piece of information for a period of time, and consolidates the outgoing change notifications, prior to sending them to one or more receiving nodes.

43. The system of claim 32, wherein the programming language constructs to represent the replicas for pieces of shared information at the node are generated from a code generator that uses an information model as its input.

44. The system of claim 35, wherein the handshake operation further comprises deleting confirmed messages from the incoming message list to maintain an unconfirmed incoming message list and deleting confirmed messages from the outgoing message list to maintain an unconfirmed outgoing message list.

45. The system of claim 32, further comprised of a security manager that restricts the responses given to incoming requests by the node.

46. The system of claim 32 further comprising a virtual file system manager unit that converts, in both directions, between the representation of the pieces of shared information by the node, and an external file system view.

52. The method of claim 51, in which the request further comprises a duration for a requested lease, and in which the response further comprises the accepted duration for the lease if the lease was granted.

53. The method of claim 52, in which the response further comprises the lease group in which the responding node has placed the lease of the first piece of information to the requesting node if the lease was granted.

54. The method of claim 53, in which the lease group is a newly created lease group, and in which the response further comprises the expiration time of the newly created lease group.

55. The method of claim 53, in which the request further comprises a requested lease group for the first piece of information.

61. The method of claim 52, in which the requesting node may request an extension to a lease obtained from the responding node for a first piece of shared information and for a certain duration, which responding node may or may not grant, responding with the accepted duration for the renewed lease if granted.

62. The method of claim 53, in which the requesting node may request an extension to the set of leases granted through a lease group by a responding node for a certain duration, which the responding node may or may not grant, responding with the accepted duration for the renewed lease group if granted.

63. The method of claim 52, in which the requesting node may request a cancellation of a lease obtained from the responding node for a replica of the first piece of shared information.

64. The method of claim 53, in which the requesting node may request a cancellation of a lease group obtained from the responding node, thereby canceling the leases of all replicas held by requesting node and subject to the said lease group.

69. The method of claim 52, wherein the expiration or cancellation of a lease causes the replicas subject to the lease to be deleted immediately at the node that had obtained the lease.

70. The method of claim 52, wherein the expiration or cancellation of a lease causes the replicas subject to the lease to become zombies at the node that had obtained the lease.

71. The method of claim 70, in which a first node holding one or more zombies generates and sends a message to a second node requesting the revival of the zombies, the message comprising the unique identifiers of the pieces of shared information whose replicas became zombies, and in which the second node may grant or reject the zombie revival request, issuing a new lease or lease group if the request was granted, and the response further comprising the serialized representation of the revived zombies.

47. A method for sharing a plurality of related pieces of information among two or more heterogeneous nodes of a distributed system, wherein the pieces of the shared information may be updated frequently by one or more of the nodes, and wherein the shared information is kept coherent, comprising:

utilizing a common communication protocol that runs on a plurality of reliable or non-reliable communication transports between nodes wherein the common communications protocol is agreed to by the nodes of the distributed system; and

sharing information among the nodes using an information model that governs the structure of the shared information when nodes communicate with each other about the shared information, that is agreed to by the nodes of the distributed system.

48. The method of claim 47, wherein the information model further comprises one or more of entity objects, relationship objects, properties of entity objects and properties of relationship objects.

49. The method of claim 47, where the communications protocol is a symmetrical protocol.

50. The method of claim 47, wherein the sharing of information further comprises exchanging a one or more unique messages between two or more nodes, wherein each unique message is identified by a unique message identifier that is incremented for each new message sent from a first node to a second node, and further comprising, detecting, at the second node, lost messages based on a missing identifier in the sequence of identifiers, so that the second node is able to determine the identifiers of, and to request the resending of lost messages from the first node.

51. The method of claim 47, in which a requesting node may request a replica of a first piece of the shared information from a responding node, the request comprising a unique identifier for the first piece of the shared information, and in which the responding node may or may not grant the request, sending a response comprising the serialized representation of the first piece of the shared information if the lease is granted.

56. The method of claim 47, in which an intermediate node may act as an intermediary to a first node for a second node, passing on any valid messages from the first node to second node, and from the second node to the first node, with or without inspection and processing of the messages.

57. The method of claim 51, in which the requesting node specifies a scope parameter that indicates which pieces of information other than the directly requested piece of information are requested to be replicated, and, if granted, in which the granting node responds with a plurality of serialized replicas reflecting the scope parameter.

58. The method of claim 57, in which the responding node responds by categorizing serialized replicas of one or more entity objects as complete, or as incomplete entities, and the response further comprising a list of identifiers for replicas of entity objects at requesting node which have now become complete as a result of the response, and the response further comprising a list of identifiers of relationship objects at the requesting node which need to be consulted to determine the correct set of relationship objects related to the now-complete set of entities.

59. The method of claim 58, in which the requesting node further generates and sends a message to the responding node requesting information that allows the requesting node to turn an incomplete replica of an entity object at the requesting node into a complete replica, and in which the responding node responds with the information, or denies to respond.

60. The method of claim 59, in which the requested and obtained information comprises serialized complete replicas of entity objects, serialized incomplete replicas of entity objects, serialized replicas of relationship objects, identifiers of entity objects replicas of which the requesting node holds that become complete as a result of receiving the response, and identifiers of relationship objects replicas of which the requesting node holds that are consulted to construct the complete replicas.

65. The method of claim 47 further comprising a first node receiving an announcement of the permanent unavailability of a second node to share information and, in response to the announcement, terminating all leases of that first node participates in with the second node and removing information held by first node about the second node.

66. The method of claim 47 further comprising a second node receiving an announcement of the temporary unavailability of a first node to share information for some expected duration and, in response to the announcement, the second node holding the outgoing messages to first node until the first node is again available.

67. The method of claim 66, wherein holding the outgoing messages further comprises consolidating the held outgoing messages syntactically or semantically in order to reduce the number of outgoing messages and the size of their information content.

68. The method of claim 47, wherein a first node generates and sends a message to a second node requesting update rights for a first piece of shared information a replica of which it holds, the replica being subject to a currently active lease between the first node and second node in either direction, and wherein the first node receives, from second node, a message in response to the request, granting or denying the request for update rights;

and further, if the request is granted, wherein the update rights to replicas of the first piece of information pass from the second node to the first node.

72. The method of claim 47 further comprising propagating updates made to any replica R of a first piece of the shared information at a first node to all nodes holding replicas of the first piece of the shared information, and said other replicas being updated to the same state as the updated replica R.

73. The method of claim 72 further comprising propagating updates along the edges of the replication graph.

74. The method of claim 72, wherein updates are updates of entity objects or updates of relationship objects, updates of a entity object being a) updates to one or more of the properties of the entity object, b) creation of relationship objects related to the entity object, c) deletion of relationship objects related to the entity object, d) updates to relationship objects related to the entity object, e) deletion of the entity object itself, and where updates of a relationship object being a) updates to one or more of the properties of the relationship object, b) deletions of an entity object related to the relationship object, c) the deletion of the relationship object itself.

75. The method of claim 74 further comprising transmogrification updates of entity objects or relationship objects.

76. The method of claim 47 in which a first node generates and sends a message to a second node asking for a resynchronization of a set of replicas that it has leased from the second node, the message comprising the unique identifiers of the pieces of shared information of which the set of replicas are replicas, and in which the second node grants or denies the resynchronization request, responding with a message comprising a serialized representation of the replicas for which the request is granted.

77. The method of claim 51, in which a first node generates and sends a message to a second node asking for the list of nodes that the second node participates in a lease with for first replica, the message comprising the unique identifier of the piece of shared information of which the first replica is a replica, and in which the second node grants or denies the request, responding with a message comprising the identifiers of all or some of the nodes that it participates in a lease with for first replica if the request is granted.

78. The method of claim 77, in which the first node modifies the replica graph by canceling a lease for the first replica of a piece of shared information that it has with the second node, and establishes a new lease for a replica of said piece of shared information with a third node, the third node'"'"'s identifier having been among the node identifiers sent back by the second node when asked for the list of nodes that the second node participates in a lease with for its replica of said piece of shared information.

79. The method of claim 47, in which a node responds to an incoming request with only some of the information it has, instead of the complete response.

80. The method of claim 79, in which the node denies an incoming lease request for a replica for a piece of shared information, for security reasons.

81. The method of claim 79, in which a node responds to an incoming lease request for a replica for a piece of shared information with only a portion of the serialized representation of the requested piece of shared information, for security reasons.

82. The method of claim 81, in which a node responds to an incoming lease request for a replica for a piece of shared information by stating that the piece of information is of a more general and less specific type than it is, for security reasons.

83. The method of claim 48, wherein a node serializes only some of the properties of a shared piece of information during communication with another node, for security reasons.

84. The method of claim 48, wherein a node uses a special value indicating “

the value is private”

when serializing a property of a shared piece of information during communication with another node, for security reasons.