# eXtremeDB Cluster Applications in C/C++

> To implement a C/C++ Cluster application requires, as a minimum, calling the following functions:

>         MCO_RET mco_cluster_init();
>         MCO_RET mco_cluster_params_init();
>         MCO_RET mco_cluster_db_open();
>         MCO_RET mco_cluster_listen();
>         MCO_RET mco_cluster_stop();

> A Cluster application can register a “quorum check callback” function to verify that the necessary minimum number of cluster nodes is present before database processing begins. If no quorum callback function is registered, the cluster runtime uses the `qrank` mechanism to verify a quorum, Also, a “notification callback” function can be registered to detect when nodes enter or leave the cluster.

> The following code snippet demonstrates a C/C++ Cluster application with quorum check and connection notification callback functions:

>         const char dbName [] = "clusterdb";

>         // Listen-thread
>         THREAD_PROC_DEFINE (ClusterListen, p)
>         {
>             mco_db_h db;
>             mco_db_connect (dbName, & db);
>             mco_cluster_listen (db);

>             mco_db_disconnect (db);
>             return 0;
>         }

>         // Quorum check callback function. It must return MCO_YES if the node can start
>         // (or continue) to work:
>         // parameters:
>         // uint2 *neighbor_ids - array of active neighbors
>         // uint2 n_neighbors - number of active neighbors in the neighbor_ids array
>         // void *param - user-defined context
>         mco_bool my_check_quorum(uint2 *neighbor_ids, uint2 n_neighbors, void *param)
>         {
>             mco_bool result = MCO_YES;
>             //    <some logic>
>             return result;
>         }

>         // Notification callback function.
>         void cluster_notifying(mco_db_h db, mco_cluster_notification_t notification_code,
>                             mco_cluster_node_info_t *node_info,
>                             void *param, void *user_context)
>         {

>             mco_cluster_node_info_t nodes_info[MCO_MAX_CLUSTER_SIZE];
>             uint2 nd, n_nodes_info = MCO_MAX_CLUSTER_SIZE;
>             printf("***NOTIFICATION*** Node %d (%s, qrank %d) %s.\n", node_info->node_id,
>             node_info->addr, node_info->qrank,
>             notification_code == MCO_CLUSTER_NOTIFY_NODE_CONNECT ?
>             "CONNECTED" : "DISCONNECTED");

>             mco_cluster_get_active_nodes(db, nodes_info, &n_nodes_info);
>             printf("List of active nodes: \n --------------------------\n");

>             for (nd = 0; nd < n_nodes_info; ++nd)
>             {
>                 printf("ID %d, Addr %s, Qrank %d\n", nodes_info[nd].node_id,
>                 nodes_info[nd].addr, nodes_info[nd].qrank);
>             }
>         }

>         int main ()
>         {
>             mco_cluster_params_t cl_params;
>             mco_cluster_node_params_t nodes [2];
>             mco_device_t dev [4];
>             unsigned int n_dev;
>             mco_db_params_t db_params;

>             // Initialize cluster runtime
>             mco_cluster_init ();

>             // Initialize cluster parameters
>             mco_cluster_params_init(&cl_params);

>             // Prepare dev, n_dev, db_params, as usual for mco_db_open_dev()

>             // Fill the information about nodes
>             strcpy (nodes[0].addr, "node1:10000")
>             strcpy (nodes[1].addr, "node2:10000");
>             nodes[0].qrank = nodes [1].qrank = 1;
>             cl_params.nodes = nodes;
>             cl_params.n_nodes = 2
>             cl_params.node_id = local_id; // on node1 this value should be 0

>             // on node2 this value should be 1, and so on

>             // set Quorum callback function
>             cl_params.check_quorum_func  = my_check_quorum;
>             cl_params.check_quorum_param = 0; // don't use the context

>             // set Notification callback function
>             cl_params.notifying_callback  = cluster_notifying;
>             cl_params.notifying_context = 0; // don't use the context

>             / * create cluster database * /
>             mco_cluster_db_open (dbName,

>                         clusterdb_get_dictionary(),
>                         dev,
>                         n_dev,
>                         &db_params,
>                         &cl_params);

>             / * Start mco_cluster_listen() thread * /
>             createThread (ClusterListen, 0, & hListenThread);

>             / * Work with the database as usual * /
>             / * Stop the cluster * /
>             mco_cluster_stop ();

>             / * Wait for the listen-thread to stop * /
>             THREAD_JOIN (hListenThread, res);

>             / * Close the database * /
>             mco_db_close ();
>         }

> Note that the ``mco_cluster_params_t`` struct (`cl_params`) passed to `mco_cluster_db_open()` contains the cluster specifications. Typically this will define the number of nodes (`n_nodes`), the integer identifier (`node_id`) of this node, and the node list, an array of `mco_cluster_node_params_t` structs (nodes) containing the node id (the array index; i.e. `nodes[0]` corresponds to `node_id=0`, `nodes[7]` - to `node_id=7` and so on), address and `qrank` of the other nodes in the cluster. It can also specify callback functions for quorum check and node notification.

> This sample uses a quorum check callback function `my_check_quorum()`. This allows the application to validate the collection of nodes online to determine if the cluster can begin processing.

> Note that when working with a cluster database, the `mco_trans_commit()` function may return error code: `MCO_E_CLUSTER_NOQUORUM`. This happens if one or more nodes go down, and the remaining nodes do not constitute a quorum.

> Also note the use of a notification callback function (`notifying_callback`); so the application will receive a connection event (type `notification_code` = `connect` / `disconnect`) with `node_info` and `user_context` (the application's context) parameters when nodes attach to and detach from the cluster.

> Note that the notification function is called in the context of a listener task. Thus it has to do whatever it does quickly, and can't perform any transactional activities or any active cluster operations. Only `mco_cluster_info()` and `mco_cluster_get_active_nodes()` are appropriate here.

> Several mco_cluster_listen() threads can be launched concurrently; remote transactions processing will be distributed between them.

> ## Adding a node to an active Cluster

> A new node can join an existing cluster by simply calling `mco_cluster_db_open()` with a unique `node_id` and the `address:port` and `node_id` of one of the running nodes. For example, suppose we have `nodeA` and `nodeB` running as a cluster with `nodeA` having `node_id = 0` and `nodeB` with `node_id = 1`. The new `nodeN` knows about `nodeA` but doesn't know about `nodeB`. It chooses `node_id = 2` and fills the `node_params` structure as following:

>         strcpy(node_params[0].addr, "nodeA:10000");
>         strcpy(node_params[1].addr, ""); // set empty address, because don't know about nodeB
>         strcpy(node_params[2].addr, "<my address and port>");

> To join the cluster `nodeN` then calls `mco_cluster_db_open()` as usual. Note that if `nodeN` were to accidentally choose `node_id = 1` (which is occupied by `nodeB`), `mco_cluster_db_open()` will return `MCO_E_CLUSTER_DUPLICATE_NODEID`. To determine the members of a cluster and their `node_id`, the `mco_cluster_discover()` function can be helpful.

> A node can also call `mco_cluster_detach()` /`mco_cluster_attach()` to leave or join the cluster. When a node joins the cluster it need only know the address of one of its neighbor nodes. It then obtains an updated node list from the cluster runtime.

> For example, consider a cluster with 4 nodes, A, B, C and D. Their node lists, `node_id` and `qrank` settings are as follows:

> > Node A : { "A", "B" }, `node_id = 0`, `qrank = 1` - Node `A` indicates its own address and also knows about node `B`. No indication of the `C` and `D` nodes presence.

> > Node B : { "A", "B"}, `node_id = 1`, `qrank - 1 -` Node `B` indicates its own address and also knows about node `A`. No indication of the `C` and `D` nodes presence.

> > Node C : { "A", "B", "C"}, `node_Id = 2`, `qrank = 0 -` Node `C` "knows" about nodes `A` and `B`. The `qrank = 0` means that disconnecting `C` does not reflect on the quorum.

> > Node D : { "A", "B", "", "D"} `node_id = 3`, `qrank = 0 -` Node `D` "knows" about nodes `A` and `B`, but not whether `C` is connected. The `qrank = 0` means that disconnecting `D` does not reflect on the quorum.

> Nodes `A` and `B` are started first, they have each other’s addresses, so they are able to communicate. Then node `C` is "attached", `C` is connected to `A` and `B` and `A` and `B` are able to add `C` into their cluster lists. Then `D` is attached. It is first connected to `A` and `B` (the addresses of which are known to `D`), receives the cluster lists from `A` and `B` (that now include `C`). It is now able to connect to `C`.

> The `mco_cluster_detach()` API disconnects the database from the cluster. The database immediately becomes available for both read and write access. Note that ``mco_cluster_detach()`` decrements the overall sum of cluster `qranks`. For example, consider a cluster consisting of two working nodes `A` and `B` with qranks of 1. If node `B` fails, then the cluster (node `A`) is stopped because the quorum is no longer there (the `qrank` sum of active nodes must be greater than half of the qranks of all nodes - and 1 is not greater than 2 / 2 = 1). However, if node `B` calls `mco_cluster_detach()`, node `A` is able to continue functioning "as a cluster", since this was a “voluntary detachment” as opposed to a network / node failure. The `qrank` of node `B` becomes 0, and the quorum is still there (1 > 1/2).

> To determine the presence of “dynamic” nodes (nodes that detach from or attach to the cluster), the `mco_cluster_get_active_nodes()` API can be called at any time. Note that before calling `mco_cluster_get_active_nodes()`, the application should call `mco_cluster_info(db,&cl_info)` which returns information including the number of active nodes (`cl_info.n_active_nodes`). This parameter is then passed to function `mco_cluster_get_active_nodes()` in order to allocate appropriate space for the `nodes_info` array.

> ## Restricting replication

> Sometimes it is beneficial to restrict automated replication to a limited set of classes. This is done by declaring some of the classes as `distributed` . The `distributed` classes are declared in the `schema` and objects of these classes don’t get distributed to the cluster automatically. Instead, each node buffers (keeps an internal list) of all inserts, modifications and deletions made to its `distributed` classes. Transactions that update the local objects don’t trigger any network-related activities.

> Periodically, the application will trigger replication of the `distributed` objects to the cluster by calling one of the following functions:

>         MCO_RET mco_cluster_scatter(mco_db_h db, uint2 *class_codes, uint2 n_class_codes,
>                         uint2 *node_ids, uint2 n_node_ids);
>         MCO_RET mco_cluster_gather(mco_db_h db, uint2 *class_codes, uint2 n_class_codes,
>                         uint2 *node_ids, uint2 n_node_ids);

> The application can scatter (push) updates made to the `distributed` objects to one or many cluster nodes, or gather (pull) updates made on one or many cluster nodes to a set of remote `distributed` objects. Both operations are atomic. If conflicts are detected, the entire scatter or gather operation is rolled back. The `mco_cluster_scatter()` function will push out updates made to its distributed tables after the last successful replication of the data. Similarly, `mco_cluster_gather()` will pull updates made to the remote “distributed” objects of interest after the last successful replication.

> For example, consider 3 nodes `A`, `B` and `C` and 2 `distributed` classes `X` and `Y` in the following scenario:

> > 1. `A` creates `x1`

> > 2. `B` creates `x2`

> > 3. `A` scatter `{X}` to `{B}` [`A` sends `x1` to `B`]

> > 4. `B` gather `{X}` from `{A,C}` [no sends because `x1` is already on `B`]

> > 5. `A` creates `y1`

> > 6. `A` scatter `{X,Y}` to `{B,C}` [`A` sends `y1` to `B` and (`x1`,`y1`) to `C`]

> > 7. `C` gather `{X,Y}` from `{A,B}` [`C` gets `x2` from `B` (`x1`,`y1` already on `C`)]

> Because Cluster in general, and the `distributed` classes in particular, allows hardware (separate nodes ) to be added to the network to increase the overall throughput. if all data is distributed automatically, the scalability is limited because of network-related overhead. The `distributed` feature allows segmenting data and processing those data segments locally (quickly). Yet the results of the local processing can be shared automatically via the `mco_cluster_scatter()` and `mco_cluster_gather()` functions.

> Another method for restricting the data being replicated is by using the `local` declaration in the `schema`. Objects of classes declared as `local` are never subject to replication – i.e. they are maintained in the local database for the individual node on which they are declared but this data is never distributed to other nodes.