Mercurial > cgi-bin > hgwebdir.cgi > VMS > C_Libraries > BestEffortMessaging

annotate LossyCom.c @ 5:95a03e431480

find changesets by author, revision, files, or words in the commit message
corrected comments and variable renaming
author Merten Sach <msach@mailbox.tu-berlin.de>
date Tue, 13 Mar 2012 19:07:49 +0100
parents 7ba5a3a6102d
children
rev   line source
msach@0 1 /*
msach@0 2 * For a detailed description see header file.
msach@0 3 */
msach@0 4
msach@4 5 #include <string.h>
msach@4 6
msach@0 7 #include "LossyCom.h"
msach@0 8 #include "VMS_Implementations/VMS_impl/vmalloc.h"
msach@0 9
msach@0 10 /*
msach@5 11 * Initializes the central exchange structure and sets all trigger counter to 0
msach@0 12 * Allocates memory to fit the number of endpoints.
msach@0 13 * Returns NULL if an error occurs.
msach@0 14 */
msach@0 15 lossyCom__exchange_t* lossyCom__initialize(uint16_t numEndpoints)
msach@0 16 {
msach@5 17 lossyCom__exchange_t* centralExchange;
msach@0 18
msach@5 19 centralExchange = VMS_WL__malloc(sizeof(lossyCom__exchange_t));
msach@5 20 if(centralExchange == NULL)
msach@0 21 return NULL;
msach@0 22
msach@5 23 centralExchange->broadcastTriggerCounter = 0;
msach@5 24 centralExchange->numEndpoints = numEndpoints;
msach@5 25 centralExchange->outboxArray = VMS_WL__malloc(sizeof(lossyCom__msg_t)*numEndpoints);
msach@5 26 if(centralExchange->outboxArray == NULL){
msach@5 27 VMS_WL__free(centralExchange);
msach@0 28 return NULL;
msach@0 29 }
msach@0 30
msach@5 31 centralExchange->p2pTriggerCounters = VMS_WL__malloc(sizeof(uint16_t)*numEndpoints);
msach@5 32 if(centralExchange->p2pTriggerCounters == NULL){
msach@5 33 VMS_WL__free(centralExchange->outboxArray);
msach@5 34 VMS_WL__free(centralExchange);
msach@4 35 return NULL;
msach@4 36 }
msach@4 37
msach@4 38 //reset all point 2 point trigger counter
msach@5 39 memset((void*)centralExchange->p2pTriggerCounters, 0, sizeof(uint16_t)*numEndpoints);
msach@4 40
msach@5 41 return centralExchange;
msach@0 42 }
msach@0 43
msach@1 44 /*
msach@1 45 * Connects the local endpoint to the central exchange structure.
msach@5 46 * This registers a message handler that handles all incoming messages.
msach@1 47 * Also an endpointID is set this ID has to be between 0 and total number of
msach@5 48 * endpoints-1 and unique.
msach@1 49 */
msach@0 50 void lossyCom__initialize_endpoint(lossyCom__endpoint_t* localEndpoint,
msach@0 51 lossyCom__exchange_t* centralExchange,
msach@0 52 lossyCom__endpointID_t endpointID,
msach@1 53 lossyCom__msgHandler msgHandler,
msach@1 54 void* msgHandlerData)
msach@0 55 {
msach@4 56 localEndpoint->lastReceivedBroadcastTrigger = 0;
msach@0 57 localEndpoint->endpointID = endpointID;
msach@1 58 localEndpoint->centralExchange = centralExchange;
msach@1 59 localEndpoint->msgHandler = msgHandler;
msach@1 60 localEndpoint->msgHandlerData = msgHandlerData;
msach@0 61 }
msach@0 62
msach@5 63 /*
msach@5 64 * Prepare a lossyCom__msg_t in the correct format that contains the trigger
msach@5 65 * value, the receiver endpoint ID and the message body
msach@5 66 */
msach@4 67 inline lossyCom__msg_t prepareMsg(uint16_t triggerValue,
msach@4 68 lossyCom__endpointID_t receiverEndpointID,
msach@5 69 lossyCom__msgBody_t msgBody)
msach@4 70 {
msach@4 71 lossyCom__msg_t msgDraft;
msach@4 72
msach@5 73 msgDraft = (0 | msgBody);
msach@4 74 msgDraft |= ((lossyCom__msg_t)receiverEndpointID << ENDPOINT_ID_SHIFT);
msach@4 75 msgDraft |= ((lossyCom__msg_t)triggerValue << TRIGGER_SHIFT);
msach@4 76
msach@4 77 return msgDraft;
msach@4 78 }
msach@4 79
msach@1 80 /*
msach@1 81 * This broadcasts a message to all connected receivers
msach@1 82 */
msach@4 83 void lossyCom__broadcastMsg(lossyCom__endpoint_t* localEndpoint,
msach@4 84 lossyCom__msgBody_t msgBody)
msach@0 85 {
msach@4 86 lossyCom__exchange_t* centralExchange = localEndpoint->centralExchange;
msach@4 87 uint16_t increasedTrigger;
msach@4 88 lossyCom__msg_t msg;
msach@4 89
msach@5 90 increasedTrigger = centralExchange->broadcastTriggerCounter +1;
msach@4 91
msach@4 92 //build message
msach@4 93 msg = prepareMsg(increasedTrigger, BROADCAST_ID, msgBody);
msach@4 94
msach@4 95 //write msg to central exchange
msach@4 96 centralExchange->outboxArray[localEndpoint->endpointID] = msg;
msach@4 97
msach@4 98 //update broadcast trigger counter
msach@5 99 centralExchange->broadcastTriggerCounter = increasedTrigger;
msach@0 100 }
msach@0 101
msach@1 102 /*
msach@1 103 * This sends a message another endpoint. Again it is not guaranteed that the
msach@1 104 * message is received. But in most cases it will.
msach@1 105 */
msach@4 106 void lossyCom__sendMsg(lossyCom__endpoint_t* localEndpoint,
msach@0 107 lossyCom__endpointID_t receiverEndpointID,
msach@4 108 lossyCom__msgBody_t msgBody)
msach@0 109 {
msach@4 110 lossyCom__exchange_t* centralExchange = localEndpoint->centralExchange;
msach@4 111 lossyCom__msg_t msg;
msach@4 112 uint16_t increasedTrigger;
msach@0 113
msach@4 114 increasedTrigger =
msach@5 115 centralExchange->p2pTriggerCounters[receiverEndpointID] +1;
msach@0 116
msach@4 117 //build message
msach@4 118 msg = prepareMsg(increasedTrigger, receiverEndpointID, msgBody);
msach@0 119
msach@0 120 //write msg to central exchange
msach@4 121 centralExchange->outboxArray[localEndpoint->endpointID] = msg;
msach@0 122
msach@4 123 //write back increased trigger counter
msach@5 124 centralExchange->p2pTriggerCounters[receiverEndpointID] = increasedTrigger;
msach@0 125 }
msach@0 126
msach@4 127 int inline isUnreceivedMsg(uint16_t msgTrigger,
msach@4 128 uint16_t lastReceivedTrigger,
msach@4 129 uint16_t triggerSnapshot)
msach@0 130 {
msach@4 131 if(msgTrigger > lastReceivedTrigger ||
msach@4 132 msgTrigger <= triggerSnapshot)
msach@4 133 {
msach@4 134 // check if the message is new (msg trigger > archived trigger)
msach@4 135 // and already valid (msgTrigger <= currentTriggerCopy)
msach@4 136 if((msgTrigger > lastReceivedTrigger &&
msach@4 137 msgTrigger <= triggerSnapshot) ||
msach@4 138 ((int64_t) triggerSnapshot- // check for triggerCounterOverflow
msach@4 139 (int64_t)lastReceivedTrigger < -MAX_TRIGGER/2))
msach@4 140 {
msach@4 141 return TRUE;
msach@4 142 }
msach@4 143 }
msach@4 144 return FALSE;
msach@4 145 }
msach@4 146
msach@4 147 void lossyCom__receiveMsg(lossyCom__endpoint_t* localEndpoint)
msach@4 148 {
msach@4 149 uint16_t broadcastTriggerSnapshot, p2pTriggerSnapshot;
msach@4 150 lossyCom__exchange_t* centralExchange;
msach@0 151 lossyCom__endpointID_t senderEndpointID;
msach@4 152 lossyCom__endpointID_t receiverEndpointID;
msach@0 153 lossyCom__msg_t msgCopy;
msach@0 154 uint16_t msgTrigger;
msach@0 155 lossyCom__msgBody_t msgBody;
msach@0 156
msach@4 157 centralExchange = localEndpoint->centralExchange;
msach@4 158 //save trigger counter to know find valid messages
msach@5 159 broadcastTriggerSnapshot = centralExchange->broadcastTriggerCounter;
msach@4 160 p2pTriggerSnapshot =
msach@5 161 centralExchange->p2pTriggerCounters[localEndpoint->endpointID];
msach@1 162
msach@0 163 //new message arrived if trigger counter is higher than the last time read
msach@4 164 if( broadcastTriggerSnapshot > localEndpoint->lastReceivedBroadcastTrigger ||
msach@4 165 p2pTriggerSnapshot > localEndpoint->lastReceivedp2pTrigger)
msach@0 166 {
msach@4 167 senderEndpointID = 0;
msach@4 168 //search outboxes for new messages
msach@4 169 while(senderEndpointID < centralExchange->numEndpoints)
msach@0 170 {
msach@4 171 //ignore own outbox
msach@2 172 if(senderEndpointID != localEndpoint->endpointID)
msach@0 173 {
msach@4 174 msgCopy = centralExchange->outboxArray[senderEndpointID];
msach@2 175 msgTrigger = 0xFFFF & (msgCopy >> TRIGGER_SHIFT);
msach@4 176 receiverEndpointID = 0xFFFF & (msgCopy >> ENDPOINT_ID_SHIFT);
msach@3 177
msach@4 178 if(receiverEndpointID == BROADCAST_ID){//receive broadcast message
msach@4 179 if(isUnreceivedMsg(msgTrigger,
msach@4 180 localEndpoint->lastReceivedBroadcastTrigger,
msach@4 181 broadcastTriggerSnapshot))
msach@2 182 {
msach@3 183 //let the message handler parse the message
msach@4 184 msgBody = 0xFFFFFFFF & msgCopy;
msach@3 185 //only receive broadcast and p2p for own receiverID
msach@4 186
msach@4 187 (*(localEndpoint->msgHandler))(senderEndpointID,
msach@4 188 msgBody,
msach@4 189 localEndpoint->msgHandlerData);
msach@4 190 }
msach@4 191 }else{//point 2 point message
msach@4 192 if(receiverEndpointID == localEndpoint->endpointID &&
msach@4 193 isUnreceivedMsg(msgTrigger,
msach@4 194 localEndpoint->lastReceivedp2pTrigger,
msach@4 195 p2pTriggerSnapshot))
msach@4 196 {
msach@4 197 //let the message handler parse the message
msach@4 198 msgBody = 0xFFFFFFFF & msgCopy;
msach@4 199 //only receive broadcast and p2p for own receiverID
msach@4 200
msach@4 201 (*(localEndpoint->msgHandler))(senderEndpointID,
msach@4 202 msgBody,
msach@4 203 localEndpoint->msgHandlerData);
msach@4 204 }
msach@2 205 }
msach@0 206 }
msach@2 207 senderEndpointID++;
msach@4 208 }//search outbox loop
msach@0 209 }
msach@2 210 //save last TriggerCounter of last parsed Msg
msach@4 211 localEndpoint->lastReceivedBroadcastTrigger = broadcastTriggerSnapshot;
msach@4 212 localEndpoint->lastReceivedp2pTrigger = p2pTriggerSnapshot;
msach@0 213 }