· 8 years ago · Feb 15, 2018, 05:52 PM
1//Global variables
2vector<Node> MP1MemList, keyReplicasList;
3size_t ringpos = 0, ringSize = 0, htSize = 0;
4string msgCUstring, msgRDstring, valEntryStr, keyStr, valStr;
5Address *prim_addr, *sec_addr, *ter_addr;
6Message *msgCU, *msgRD;
7Entry *valEntry;
8int foundcountMP2 = 0, upreplyCount = 0, delreplyCount = 0, upqrmCount = 0, delqrmCount = 0, uptxnCount = 0, deltxnCount = 0, qrmCount = 0, txnCount = 0, replyCount = 0;
9int crreplyCount = 0, crtxnCount = 0;
10vector<Message>::iterator foundposMP2;
11vector<Message>::iterator Itr;
12vector<int>::iterator uniqueItr;
13map<string, string>::iterator hashItr;
14vector<int>::iterator rsItr;
15tuple<int, bool> replyRec(0, false);
16list<tuple<int, bool>>::iterator txnItr;
17list<int> txnIDList, uniquetxnIDList;
18MessageType msgType;
19
20/**
21 * constructor
22 */
23MP2Node::MP2Node(Member *memberNode, Params *par, EmulNet *emulNet, Log *log, Address *address)
24{
25 this->memberNode = memberNode;
26 this->par = par;
27 this->emulNet = emulNet;
28 this->log = log;
29 ht = new HashTable();
30 this->memberNode->addr = *address;
31 TXNID.clear();
32 msgBox.clear();
33 ringSizeList.clear();
34}
35
36/**
37 * Destructor
38 */
39MP2Node::~MP2Node()
40{
41 delete ht;
42 delete memberNode;
43}
44
45/**
46 * FUNCTION NAME: updateRing
47 *
48 * DESCRIPTION: This function does the following:
49 * 1) Gets the current membership list from the Membership Protocol (MP1Node)
50 * The membership list is returned as a vector of Nodes. See Node class in Node.h
51 * 2) Constructs the ring based on the membership list
52 * 3) Calls the Stabilization Protocol
53 */
54void MP2Node::updateRing()
55{
56 /*
57 * Implement this. Parts of it are already implemented
58 */
59
60 bool change = false;
61 ringpos = 0;
62
63 /*
64 * Step 1. Get the current membership list from Membership Protocol / MP1
65 */
66 MP1MemList = getMembershipList();
67
68 /*
69 * Step 2: Construct the ring
70 */
71 // Sort the list based on the hashCode
72 if (!MP1MemList.empty())
73 {
74 sort(MP1MemList.begin(), MP1MemList.end());
75
76 //create CHORD ring in ascending order of hash code clockwise
77 this->ring = MP1MemList;
78
79 //store ring sizes in history
80 ringSize = ring.size();
81 ringSizeList.emplace_back(ringSize);
82 }
83
84 int n = ringSizeList.size();
85 if (n > 1)
86 {
87 if (ringSizeList[n - 1] != ringSizeList[n - 2])
88 change = true;
89 }
90
91 /*
92 * Step 3: Run the stabilization protocol IF REQUIRED (bool change = true)
93 */
94 // Run stabilization protocol if the hash table size is greater than zero and if there has been a change in the ring
95
96 if ((ht->currentSize() > 0) && (change == true))
97 runstabilizationProtocol();
98}
99
100/**
101 * FUNCTION NAME: getMembershipList
102 *
103 * DESCRIPTION: This function goes through the membership list from the Membership protocol/MP1 and
104 * i) generates the hash code for each member
105 * ii) populates the ring member in MP2Node class
106 * It returns a vector of Nodes. Each element in the vector contain the following fields:
107 * a) Address of the node
108 * b) Hash code obtained by consistent hashing of the Address
109 */
110vector<Node> MP2Node::getMembershipList()
111{
112
113 MP1MemList.clear();
114 Address *addressOfThisMember = new Address();
115
116 for (this->memberNode->myPos = memberNode->memberList.begin(); this->memberNode->myPos != this->memberNode->memberList.end(); ++this->memberNode->myPos)
117 {
118
119 addressOfThisMember->addr[0] = (*memberNode->myPos).getid();
120 addressOfThisMember->addr[4] = (*memberNode->myPos).getport();
121 MP1MemList.emplace_back(Node(*addressOfThisMember));
122 }
123
124 delete addressOfThisMember;
125 return MP1MemList;
126}
127
128/**
129 * FUNCTION NAME: hashFunction
130 *
131 * DESCRIPTION: This functions hashes the key and returns the position on the ring
132 * HASH FUNCTION USED FOR CONSISTENT HASHING
133 *
134 * RETURNS:
135 * size_t position on the ring
136 */
137size_t MP2Node::hashFunction(string key)
138{
139 std::hash<string> hashFunc;
140 size_t ret = hashFunc(key);
141 return ret % RING_SIZE;
142}
143
144//-----------------------------------------CLIENT CRUD API-------------------------------------------------------------
145
146/**
147 * FUNCTION NAME: clientCreate
148 *
149 * DESCRIPTION: client side CREATE API
150 * The function does the following:
151 * 1) Constructs the message
152 * 2) Finds the replicas of this key
153 * 3) Sends a message to the replica
154 */
155void MP2Node::clientCreate(string key, string value)
156{
157 /*
158 * Implement this
159 */
160
161 ringpos = hashFunction(key); //hashing key to get a ring position
162
163 //Identify Primary, Scondary, Tertiary Nodes and addresses to send message to
164
165 if (ring.size() >= 3)
166 {
167 // if pos <= min || pos > max, the leader is the min
168 if (ringpos <= ring.at(0).getHashCode() || ringpos > ring.at(ring.size() - 1).getHashCode())
169 {
170 prim_addr = ring.at(0).getAddress();
171 sec_addr = ring.at(1).getAddress();
172 ter_addr = ring.at(2).getAddress();
173 }
174 else
175 {
176 // go through the ring until pos <= node
177
178 for (int i = 1; i < ring.size(); i++)
179 {
180
181 if (ringpos <= ring.at(i).getHashCode())
182 {
183 prim_addr = ring.at(i % ring.size()).getAddress();
184 sec_addr = ring.at((i + 1) % ring.size()).getAddress();
185 ter_addr = ring.at((i + 2) % ring.size()).getAddress();
186 break;
187 }
188 }
189 }
190
191 //send CREATE messages via emulNet to PRIM, SEC, TER replica nodes
192
193 g_transID += 1;
194
195 msgCU = new Message(g_transID, memberNode->addr, CREATE, key, value, PRIMARY);
196 msgCUstring = msgCU->toString();
197 emulNet->ENsend(&memberNode->addr, prim_addr, msgCUstring);
198 msgBox.emplace_back(*msgCU);
199 delete msgCU;
200 msgCUstring.clear();
201
202 msgCU = new Message(g_transID, memberNode->addr, CREATE, key, value, SECONDARY);
203 msgCUstring = msgCU->toString();
204 emulNet->ENsend(&memberNode->addr, sec_addr, msgCUstring);
205 msgBox.emplace_back(*msgCU);
206 delete msgCU;
207 msgCUstring.clear();
208
209 msgCU = new Message(g_transID, memberNode->addr, CREATE, key, value, TERTIARY);
210 msgCUstring = msgCU->toString();
211 emulNet->ENsend(&memberNode->addr, ter_addr, msgCUstring);
212 msgBox.emplace_back(*msgCU);
213 delete msgCU;
214 msgCUstring.clear();
215 }
216 else
217 {
218 cout << "ring size is less than 3. Exiting...";
219 exit(1);
220 }
221}
222
223/**
224 * FUNCTION NAME: clientRead
225 *
226 * DESCRIPTION: client side READ API
227 * The function does the following:
228 * 1) Constructs the message
229 * 2) Finds the replicas of this key
230 * 3) Sends a message to the replica
231 */
232void MP2Node::clientRead(string key)
233{
234 /*
235 * Implement this
236 */
237
238 //find the nodes where the replicas with they key to be read are stored
239 keyReplicasList.clear();
240 keyReplicasList = findNodes(key);
241 g_transID += 1;
242 int kr = keyReplicasList.size();
243
244 //send READ messages via emulNet to all replica nodes where key is present
245 if (kr >= 3)
246 {
247
248 for (int p = 0; p < kr; p++)
249 {
250
251 msgRD = new Message(g_transID, memberNode->addr, READ, key);
252 msgRDstring = msgRD->toString();
253 emulNet->ENsend(&memberNode->addr, keyReplicasList[p].getAddress(), msgRDstring);
254 msgBox.emplace_back(*msgRD);
255 delete msgRD;
256 msgRDstring.clear();
257 }
258 }
259
260 else
261 cout << "error: 3 replicas not found to send msgs to READ";
262}
263
264/**
265 * FUNCTION NAME: clientUpdate
266 *
267 * DESCRIPTION: client side UPDATE API
268 * The function does the following:
269 * 1) Constructs the message
270 * 2) Finds the replicas of this key
271 * 3) Sends a message to the replica
272 */
273void MP2Node::clientUpdate(string key, string value)
274{
275 /*
276 * Implement this
277 */
278
279 //find the Replicas where the keys are stored
280 keyReplicasList.clear();
281 keyReplicasList = findNodes(key);
282 int ku = keyReplicasList.size();
283
284 //send update messages to all the replicas in the keyReplicasList to update with the new value
285
286 if (ku >= 3)
287 {
288
289 g_transID += 1;
290
291 msgCU = new Message(g_transID, memberNode->addr, UPDATE, key, value, PRIMARY);
292 msgCU->success = false;
293 msgCUstring = msgCU->toString();
294 emulNet->ENsend(&memberNode->addr, keyReplicasList[0].getAddress(), msgCUstring);
295 msgBox.emplace_back(*msgCU);
296 delete msgCU;
297 msgCUstring.clear();
298
299 msgCU = new Message(g_transID, memberNode->addr, UPDATE, key, value, SECONDARY);
300 msgCU->success = false;
301 msgCUstring = msgCU->toString();
302 emulNet->ENsend(&memberNode->addr, keyReplicasList[1].getAddress(), msgCUstring);
303 msgBox.emplace_back(*msgCU);
304 delete msgCU;
305 msgCUstring.clear();
306
307 msgCU = new Message(g_transID, memberNode->addr, UPDATE, key, value, TERTIARY);
308 msgCU->success = false;
309 msgCUstring = msgCU->toString();
310 emulNet->ENsend(&memberNode->addr, keyReplicasList[2].getAddress(), msgCUstring);
311 msgBox.emplace_back(*msgCU);
312 delete msgCU;
313 msgCUstring.clear();
314 }
315
316 else
317 cout << "error: 3 replicas not found to send msgs to UPDATE";
318}
319
320/**
321 * FUNCTION NAME: clientDelete
322 *
323 * DESCRIPTION: client side DELETE API
324 * The function does the following:
325 * 1) Constructs the message
326 * 2) Finds the replicas of this key
327 * 3) Sends a message to the replica
328 */
329void MP2Node::clientDelete(string key)
330{
331 /*
332 * Implement this
333 *
334 */
335
336 //find the nodes where the replicas with the key to be deleted are stored
337 keyReplicasList.clear();
338 keyReplicasList = findNodes(key);
339 int kd = keyReplicasList.size();
340 g_transID += 1;
341
342 //send DELETE messages via emulNet to all replica nodes where key is present
343 if (kd >= 3)
344 {
345
346 for (int j = 0; j < kd; j++)
347 {
348
349 msgRD = new Message(g_transID, memberNode->addr, DELETE, key);
350 msgRD->success = false;
351 msgRDstring = msgRD->toString();
352 emulNet->ENsend(&memberNode->addr, keyReplicasList[j].getAddress(), msgRDstring);
353 msgBox.emplace_back(*msgRD);
354 delete msgRD;
355 msgRDstring.clear();
356 }
357 }
358
359 else
360 cout << "error: 3 replicas not found to send msgs to DELETE";
361}
362
363/**
364 * FUNCTION NAME: findNodes
365 *
366 * DESCRIPTION: Get the replicas of the given keyfunction
367 * This function is responsible for finding the replicas of a key
368 */
369
370vector<Node> MP2Node::findNodes(string key)
371{
372
373 size_t pos = hashFunction(key);
374 vector<Node> addr_vec;
375
376 if (ring.size() >= 3)
377 {
378 // if pos <= min || pos > max, the leader is the min
379 if (pos <= ring.at(0).getHashCode() || pos > ring.at(ring.size() - 1).getHashCode())
380 {
381 addr_vec.emplace_back(ring.at(0));
382 addr_vec.emplace_back(ring.at(1));
383 addr_vec.emplace_back(ring.at(2));
384 }
385 else
386 {
387 // go through the ring until pos <= node
388 for (int i = 1; i < ring.size(); i++)
389 {
390 Node addr = ring.at(i);
391 if (pos <= addr.getHashCode())
392 {
393 addr_vec.emplace_back(addr);
394 addr_vec.emplace_back(ring.at((i + 1) % ring.size()));
395 addr_vec.emplace_back(ring.at((i + 2) % ring.size()));
396 break;
397 }
398 }
399 }
400 }
401
402 return addr_vec;
403}
404
405//--------------------------------------------SERVER SIDE API--------------------------------------------------------------
406/**
407 * FUNCTION NAME: createKeyValue
408 *
409 * DESCRIPTION: Server side CREATE API
410 * The function does the following:
411 * 1) Inserts key value into the local hash table
412 * 2) Return true or false based on success or failure
413 */
414bool MP2Node::createKeyValue(string key, string value, ReplicaType replica)
415{
416 /*
417 * Implement this
418 */
419 // Insert key, value, replicaType into the hash table
420
421 if (memberNode->bFailed)
422 return false;
423
424 valEntry = new Entry(value, par->getcurrtime(), replica);
425 valEntryStr = valEntry->convertToString();
426 delete valEntry;
427
428 if (ht->create(key, valEntryStr))
429 return true;
430
431 else
432 return false;
433}
434
435/**
436 * FUNCTION NAME: readKey
437 *
438 * DESCRIPTION: Server side READ API
439 * This function does the following:
440 * 1) Read key from local hash table
441 * 2) Return value
442 */
443string MP2Node::readKey(string key)
444{
445 /*
446 * Implement this
447 */
448 // Read key from local hash table and return value
449 if (memberNode->bFailed)
450 return "";
451 return ht->read(key);
452}
453
454/**
455 * FUNCTION NAME: updateKeyValue
456 *
457 * DESCRIPTION: Server side UPDATE API
458 * This function does the following:
459 * 1) Update the key to the new value in the local hash table
460 * 2) Return true or false based on success or failure
461 */
462bool MP2Node::updateKeyValue(string key, string value, ReplicaType replica)
463{
464 /*
465 * Implement this
466 */
467 // Update key in local hash table and return true or false
468 if (memberNode->bFailed)
469 return false;
470
471 valEntryStr.clear();
472 valEntry = new Entry(value, par->getcurrtime(), replica);
473 valEntryStr = valEntry->convertToString();
474 delete valEntry;
475
476 if (ht->update(key, valEntryStr))
477 return true;
478 else
479 return false;
480}
481
482/**
483 * FUNCTION NAME: deleteKey
484 *
485 * DESCRIPTION: Server side DELETE API
486 * This function does the following:
487 * 1) Delete the key from the local hash table
488 * 2) Return true or false based on success or failure
489 */
490bool MP2Node::deletekey(string key)
491{
492 /*
493 * Implement this
494 */
495 // Delete the key from the local hash table
496 if (memberNode->bFailed)
497 return false;
498
499 if (ht->deleteKey(key))
500 return true;
501 else
502 return false;
503}
504
505//---------------------------------------------------ALL NODES CHECK MESSAGES AND ACT------------------------------------------------
506
507/**
508 * FUNCTION NAME: checkMessages
509 *
510 * DESCRIPTION: This function is the message handler of this node.
511 * This function does the following:
512 * 1) Pops messages from the queue
513 * 2) Handles the messages according to message types
514 */
515void MP2Node::checkMessages()
516{
517 /*
518 * Implement this. Parts of it are already implemented
519 */
520 char *data;
521 int size;
522
523 /*
524 * Declare your local variables here
525 */
526
527 Message *msgQobj, *msgReplyObj;
528 Entry *msgRRobj;
529 string msgReplyString;
530 string readValue;
531
532 // dequeue all messages and handle them
533 while (!memberNode->mp2q.empty())
534 {
535 /*
536 * Pop a message from the queue
537 */
538 data = (char *)memberNode->mp2q.front().elt;
539 size = memberNode->mp2q.front().size;
540 memberNode->mp2q.pop();
541
542 string message(data, data + size);
543
544 /*
545 * Handle the message types here
546 */
547
548 //convert string message to object message msgQobj
549 msgQobj = new Message(message);
550 auto searchID = msgQobj->transID;
551 auto rType = msgQobj->replica;
552 MessageType mType = msgQobj->type;
553
554 //depending on message type, handle messages
555 switch (mType)
556 {
557
558 case CREATE:
559
560 if (createKeyValue(msgQobj->key, msgQobj->value, msgQobj->replica))
561 {
562 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, REPLY, true);
563 log->logCreateSuccess(&memberNode->addr, false, msgQobj->transID, msgQobj->key, msgQobj->value);
564 }
565 else
566 {
567 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, REPLY, false);
568 log->logCreateFail(&memberNode->addr, false, msgQobj->transID, msgQobj->key, msgQobj->value);
569 }
570 msgReplyString = msgReplyObj->toString();
571 emulNet->ENsend(&memberNode->addr, &msgQobj->fromAddr, msgReplyString);
572 msgBox.emplace_back(*msgReplyObj);
573
574 break;
575
576 case UPDATE:
577
578 if (updateKeyValue(msgQobj->key, msgQobj->value, msgQobj->replica))
579 {
580 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, REPLY, true);
581 log->logUpdateSuccess(&memberNode->addr, false, msgQobj->transID, msgQobj->key, msgQobj->value);
582 }
583 else
584 {
585 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, REPLY, false);
586 log->logUpdateFail(&memberNode->addr, false, msgQobj->transID, msgQobj->key, msgQobj->value);
587 }
588 msgReplyString = msgReplyObj->toString();
589 emulNet->ENsend(&memberNode->addr, &msgQobj->fromAddr, msgReplyString);
590 msgBox.emplace_back(*msgReplyObj);
591
592 break;
593
594 case READ:
595 readValue = readKey(msgQobj->key);
596 if (!readValue.empty())
597 {
598 msgRRobj = (Entry *)new Entry(readValue);
599 log->logReadSuccess(&memberNode->addr, false, msgQobj->transID, msgQobj->key, msgRRobj->value);
600 }
601
602 else
603 {
604 msgRRobj->value = readValue;
605 log->logReadFail(&memberNode->addr, false, msgQobj->transID, msgQobj->key);
606 }
607
608 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, msgRRobj->value);
609 msgReplyString = msgReplyObj->toString();
610 emulNet->ENsend(&memberNode->addr, &msgQobj->fromAddr, msgReplyString);
611 msgBox.emplace_back(*msgReplyObj);
612 break;
613
614 case DELETE:
615 if (deletekey(msgQobj->key))
616 {
617 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, REPLY, true);
618 log->logDeleteSuccess(&memberNode->addr, false, msgQobj->transID, msgQobj->key);
619 }
620 else
621 {
622 msgReplyObj = new Message(msgQobj->transID, memberNode->addr, REPLY, false);
623 log->logDeleteFail(&memberNode->addr, false, msgQobj->transID, msgQobj->key);
624 }
625 msgReplyString = msgReplyObj->toString();
626 emulNet->ENsend(&memberNode->addr, &msgQobj->fromAddr, msgReplyString);
627 msgBox.emplace_back(*msgReplyObj);
628 break;
629
630 case REPLY:
631
632 //handle REPLY from CREATE transactions
633 //finding txn count to check if all client CREATE txns sent to all 3 replicas for this reply txnID
634
635 crtxnCount = count_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == CREATE); });
636
637 //find key for this REPLY transaction
638 if ((crtxnCount > 0) && (crtxnCount % 3 == 0))
639 {
640 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == CREATE); });
641 keyStr = foundposMP2->key;
642 valStr = foundposMP2->value;
643
644 //find the addresses of all the replicas UPDATE has been sent to; replies come from these replica addresses
645 keyReplicasList.clear();
646 keyReplicasList = findNodes(keyStr);
647
648 //record all replies
649
650 if (msgQobj->fromAddr == keyReplicasList[0].nodeAddress)
651 {
652 rType = PRIMARY;
653 replyRec = make_tuple(searchID, true);
654 TXNID.emplace_back(replyRec);
655 replyRec = make_tuple(0, false);
656 }
657 else if (msgQobj->fromAddr == keyReplicasList[1].nodeAddress)
658 {
659 rType = SECONDARY;
660 replyRec = make_tuple(searchID, true);
661 TXNID.emplace_back(replyRec);
662 replyRec = make_tuple(0, false);
663 }
664 else if (msgQobj->fromAddr == keyReplicasList[2].nodeAddress)
665 {
666 rType = TERTIARY;
667 replyRec = make_tuple(searchID, true);
668 TXNID.emplace_back(replyRec);
669 replyRec = make_tuple(0, false);
670 }
671
672 foundcountMP2 = count_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.type == CREATE); });
673
674 if (foundcountMP2 > 0)
675 {
676 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.type == CREATE); });
677 foundposMP2->success = msgQobj->success;
678 }
679 }
680
681 //handle REPLY from UPDATE transactions
682 //finding txn count to check if all client UPDATE txns sent to all 3 replicas for this reply txnID
683
684 uptxnCount = count_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == UPDATE); });
685
686 //find key for this REPLY transaction
687 if ((uptxnCount > 0) && (uptxnCount % 3 == 0))
688 {
689 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == UPDATE); });
690 keyStr = foundposMP2->key;
691 valStr = foundposMP2->value;
692
693 //find the addresses of all the replicas UPDATE has been sent to; replies come from these replica addresses
694 keyReplicasList.clear();
695 keyReplicasList = findNodes(keyStr);
696
697 //record all replies
698
699 if (msgQobj->fromAddr == keyReplicasList[0].nodeAddress)
700 {
701 rType = PRIMARY;
702 replyRec = make_tuple(searchID, true);
703 TXNID.emplace_back(replyRec);
704 replyRec = make_tuple(0, false);
705 }
706 else if (msgQobj->fromAddr == keyReplicasList[1].nodeAddress)
707 {
708 rType = SECONDARY;
709 replyRec = make_tuple(searchID, true);
710 TXNID.emplace_back(replyRec);
711 replyRec = make_tuple(0, false);
712 }
713 else if (msgQobj->fromAddr == keyReplicasList[2].nodeAddress)
714 {
715 rType = TERTIARY;
716 replyRec = make_tuple(searchID, true);
717 TXNID.emplace_back(replyRec);
718 replyRec = make_tuple(0, false);
719 }
720
721 foundcountMP2 = count_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.type == UPDATE); });
722
723 if (foundcountMP2 > 0)
724 {
725 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.type == UPDATE); });
726 foundposMP2->success = msgQobj->success;
727 }
728 }
729 //repeat steps for DELETE reply
730
731 //finding txn count to check if all client DELETE txns sent to all 3 replicas
732
733 deltxnCount = count_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == DELETE); });
734
735 if ((deltxnCount > 0) && (deltxnCount % 3 == 0))
736 {
737
738 //find key for this REPLY transaction
739 foundcountMP2 = count_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.success == false && msgList.type == DELETE); });
740 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.success == false && msgList.type == DELETE); });
741 keyStr = foundposMP2->key;
742
743 //find the addresses of all the replicas DELETE has been sent to; replies come from these replica addresses
744 keyReplicasList.clear();
745 keyReplicasList = findNodes(keyStr);
746
747 if (msgQobj->fromAddr == keyReplicasList[0].nodeAddress)
748 {
749
750 rType = PRIMARY;
751 foundposMP2->replica = rType;
752 replyRec = make_tuple(searchID, true);
753 TXNID.emplace_back(replyRec);
754 replyRec = make_tuple(0, false);
755 }
756 else if (msgQobj->fromAddr == keyReplicasList[1].nodeAddress)
757 {
758
759 rType = SECONDARY;
760 foundposMP2->replica = rType;
761 replyRec = make_tuple(searchID, true);
762 TXNID.emplace_back(replyRec);
763 replyRec = make_tuple(0, false);
764 }
765 else if (msgQobj->fromAddr == keyReplicasList[2].nodeAddress)
766 {
767
768 rType = TERTIARY;
769 foundposMP2->replica = rType;
770 replyRec = make_tuple(searchID, true);
771 TXNID.emplace_back(replyRec);
772 replyRec = make_tuple(0, false);
773 }
774
775 foundcountMP2 = count_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.type == DELETE); });
776
777 if (foundcountMP2 > 0)
778 {
779
780 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.success == false && msgList.type == DELETE); });
781 foundposMP2->success = msgQobj->success;
782 }
783 }
784
785 break;
786
787 case READREPLY:
788
789 //finding txns to see if all client READ txns are sent to all 3 replicas
790 txnCount = count_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return msgList.transID == searchID && msgList.type == READ; });
791
792 if ((txnCount > 0) && (txnCount % 3 == 0))
793 {
794 //find key for this READREPLY transaction
795 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == READ); });
796 if (foundposMP2 != msgBox.end())
797 keyStr = foundposMP2->key;
798
799 //find the addresses of all the replicas READ has been sent to; replies come from these replica addresses
800 keyReplicasList.clear();
801 keyReplicasList = findNodes(keyStr);
802
803 if (msgQobj->fromAddr == keyReplicasList[0].nodeAddress)
804 {
805 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.success == false && msgList.type == READ); });
806 rType = PRIMARY;
807 foundposMP2->replica = rType;
808 foundposMP2->value = msgQobj->value;
809 if (foundposMP2->value != "")
810 foundposMP2->success = true;
811 else
812 foundposMP2->success = false;
813 replyRec = make_tuple(searchID, true);
814 TXNID.emplace_back(replyRec);
815 replyRec = make_tuple(0, false);
816 }
817 else if (msgQobj->fromAddr == keyReplicasList[1].nodeAddress)
818 {
819 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.success == false && msgList.type == READ); });
820 rType = SECONDARY;
821 foundposMP2->replica = rType;
822 foundposMP2->value = msgQobj->value;
823 if (foundposMP2->value != "")
824 foundposMP2->success = true;
825 else
826 foundposMP2->success = false;
827 replyRec = make_tuple(searchID, true);
828 TXNID.emplace_back(replyRec);
829 replyRec = make_tuple(0, false);
830 }
831 else if (msgQobj->fromAddr == keyReplicasList[2].nodeAddress)
832 {
833 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.success == false && msgList.type == READ); });
834 rType = TERTIARY;
835 foundposMP2->replica = rType;
836 foundposMP2->value = msgQobj->value;
837 if (foundposMP2->value != "")
838 foundposMP2->success = true;
839 else
840 foundposMP2->success = false;
841 replyRec = make_tuple(searchID, true);
842 TXNID.emplace_back(replyRec);
843 replyRec = make_tuple(0, false);
844 }
845
846 foundcountMP2 = count_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.type == READ); });
847
848 if (foundcountMP2 > 0)
849 {
850
851 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [searchID, rType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.replica == rType && msgList.success == false && msgList.type == READ); });
852 foundposMP2->success = msgQobj->success;
853 }
854 }
855
856 break;
857 }
858
859 delete msgQobj;
860 }
861
862 /*
863 * This function should also ensure all READ and UPDATE operation
864 * get QUORUM replies. Implemented here.
865 */
866
867 //create a list of transaction IDs from TXNID list (a list that captures only transactions that
868 //have received REPLY messages)
869 txnIDList.clear();
870 if (!TXNID.empty())
871 {
872
873 for (txnItr = TXNID.begin(); txnItr != TXNID.end(); txnItr++)
874 {
875
876 txnIDList.emplace_back(get<0>(*txnItr));
877 }
878 //make unique txnIDs list for iteration
879 txnIDList.sort();
880 txnIDList.unique();
881 }
882
883 while (!txnIDList.empty())
884 {
885 auto frontID = txnIDList.front();
886
887 //find key and value for this txnID
888 foundposMP2 = find_if(msgBox.begin(), msgBox.end(), [frontID](Message msgList) mutable -> bool { return (msgList.transID == frontID); });
889 keyStr = foundposMP2->key;
890 valStr = foundposMP2->value;
891 msgType = foundposMP2->type;
892
893 //determine CREATE, READ, UPDATE, DELETE transaction quorums
894
895 if (msgType == CREATE)
896 {
897 qrmCount = count_if(msgBox.begin(), msgBox.end(), [frontID](Message msgList) mutable -> bool { return (msgList.transID == frontID && msgList.success == true && msgList.type == CREATE); });
898 if (qrmCount >= 2)
899 log->logCreateSuccess(&memberNode->addr, true, frontID, keyStr, valStr);
900 else
901 log->logCreateFail(&memberNode->addr, true, frontID, keyStr, valStr);
902
903 txnIDList.pop_front();
904 TXNID.sort();
905 TXNID.remove(make_tuple(frontID, true));
906 }
907
908 if (msgType == READ)
909 {
910 qrmCount = count_if(msgBox.begin(), msgBox.end(), [frontID](Message msgList) mutable -> bool { return (msgList.transID == frontID && msgList.success == true && msgList.type == READ); });
911 if (qrmCount >= 2)
912 log->logReadSuccess(&memberNode->addr, true, frontID, keyStr, valStr);
913 else
914 log->logReadFail(&memberNode->addr, true, frontID, keyStr);
915
916 txnIDList.pop_front();
917 TXNID.sort();
918 TXNID.remove(make_tuple(frontID, true));
919 }
920
921 if (msgType == UPDATE)
922 {
923 qrmCount = count_if(msgBox.begin(), msgBox.end(), [frontID](Message msgList) mutable -> bool { return (msgList.transID == frontID && msgList.success == true && msgList.type == UPDATE); });
924 if (qrmCount >= 2)
925 log->logUpdateSuccess(&memberNode->addr, true, frontID, keyStr, valStr);
926 else
927 log->logUpdateFail(&memberNode->addr, true, frontID, keyStr, valStr);
928
929 txnIDList.pop_front();
930 TXNID.sort();
931 TXNID.remove(make_tuple(frontID, true));
932 }
933
934 if (msgType == DELETE)
935 {
936 qrmCount = count_if(msgBox.begin(), msgBox.end(), [frontID](Message msgList) mutable -> bool { return (msgList.transID == frontID && msgList.success == true && msgList.type == DELETE); });
937 if (qrmCount >= 2)
938 log->logDeleteSuccess(&memberNode->addr, true, frontID, keyStr);
939 else
940 log->logDeleteFail(&memberNode->addr, true, frontID, keyStr);
941
942 txnIDList.pop_front();
943 TXNID.sort();
944 TXNID.remove(make_tuple(frontID, true));
945 }
946 }
947}
948
949/**
950 * FUNCTION NAME: recvLoop
951 *
952 * DESCRIPTION: Receive messages from EmulNet and push into the queue (mp2q)
953 */
954bool MP2Node::recvLoop()
955{
956 if (memberNode->bFailed)
957 {
958 return false;
959 }
960 else
961 {
962 return emulNet->ENrecv(&(memberNode->addr), this->enqueueWrapper, NULL, 1, &(memberNode->mp2q));
963 }
964}
965
966/**
967 * FUNCTION NAME: enqueueWrapper
968 *
969 * DESCRIPTION: Enqueue the message from Emulnet into the queue of MP2Node
970 */
971int MP2Node::enqueueWrapper(void *env, char *buff, int size)
972{
973 Queue q;
974 return q.enqueue((queue<q_elt> *)env, (void *)buff, size);
975}
976/**
977 * FUNCTION NAME: stabilizationProtocol
978 *
979 * DESCRIPTION: This runs the stabilization protocol in case of Node joins and leaves
980 * It ensures that there always 3 copies of all keys in the DHT at all times
981 * The function does the following:
982 * 1) Ensures that there are three "CORRECT" replicas of all the keys in spite of failures and joins
983 * Note:- "CORRECT" replicas implies that every key is replicated in its two neighboring nodes in the ring
984 */
985void MP2Node::runstabilizationProtocol()
986{
987 /*
988 * Implement this
989 */
990
991 if (!memberNode->bFailed)
992 {
993 //find size of hash table for the node and iterate through all the keys to recreate 3 replicas
994 htSize = ht->currentSize();
995
996 for (hashItr = ht->hashTable.begin(); hashItr != ht->hashTable.end(); hashItr++)
997 {
998 keyStr = hashItr->first;
999 valEntry = new Entry(hashItr->second);
1000 valStr = valEntry->value;
1001 clientCreate(keyStr, valStr);
1002 }
1003
1004 delete valEntry;
1005 }
1006}
1007
1008/**
1009 * FUNCTION NAME: findMLEntryPos
1010 *
1011 * DESCRIPTION: finding if an ID exists in member list
1012 *
1013 */
1014
1015int MP2Node::foundID(vector<Message> searchMsg, int searchID, MessageType mType)
1016{
1017 int fcount = 0;
1018 fcount = count_if(searchMsg.begin(), searchMsg.end(), [searchID, mType](Message msgList) mutable -> bool { return (msgList.transID == searchID && msgList.type == mType); });
1019 return fcount;
1020}