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

annotate LossyCom.c @ 4:7ba5a3a6102d

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