MultiThreading--Token-Bucket/tokenGeneratingThread.c at master · surajhes/MultiThreading--Token-Bucket · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
//
// Created by surajhs04 on 9/18/16.
//
#include <time.h>
#include <sys/types.h>
#include <pthread.h>
#include <sys/time.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <signal.h>
#include "definitions.h"

int tokenBucketCount = 0;

int stopService;

int stopToken = 0;

long double simulationStartTime;

My402List queue1;

My402List queue2;

struct Statistics statistics;

long double getCurrentTimeMS();

void *generateToken(void *arg) {
    sigemptyset(&set);
    sigaddset(&set, SIGINT);
    sigprocmask(SIG_BLOCK, &set, 0);
    struct emulationParameters *parameters = (struct emulationParameters *) arg;
    int bucketSize = parameters->B;
    long double tokenThreadWakeUpTime = 0.0, tokenThreadServiceTime = 0, currentTimeInMilliSeconds;
    long double actualRate = (1 / parameters->r) * 1000; //in ms
    if ((actualRate / 1000) > 10) {
        actualRate = 10000.0;
    }
    long double sleepTime = actualRate - (tokenThreadServiceTime - tokenThreadWakeUpTime); //in ms
    long double delta = 0;
    long double interTokenArrivalRate = getCurrentTimeMS();
    int tokenCount = 0;
    while (1) {
        if (sleepTime > 0)
            usleep((useconds_t) (sleepTime * 1000));
        tokenThreadWakeUpTime = getCurrentTimeMS();
        interTokenArrivalRate = tokenThreadWakeUpTime - interTokenArrivalRate;
        pthread_mutex_lock(&mutex);
        if (My402ListEmpty(&queue1) && stopToken) {
            stopService = 1;
            pthread_cond_broadcast(&queueNotEmpty);
            pthread_mutex_unlock(&mutex);
            break;
        }
        tokenCount++;
        tokenBucketCount++;
        statistics.totalTokensGenerated++;
        if (tokenBucketCount < 2) {
            if (tokenBucketCount > bucketSize) {
                statistics.totalDroppedTokens++;
                fprintf(stdout, "\n%012.3Lfms: token t%d arrives,dropped",
                        getCurrentTimeMS() - simulationStartTime,
                        tokenCount);
                tokenBucketCount--;
            }
            fprintf(stdout, "\n%012.3Lfms: token t%d arrives,token bucket now has %d token",
                    getCurrentTimeMS() - simulationStartTime,
                    tokenCount, tokenBucketCount);
        } else {
            if (tokenBucketCount > bucketSize) {
                statistics.totalDroppedTokens++;
                fprintf(stdout, "\n%012.3Lfms: token t%d arrives,dropped",
                        getCurrentTimeMS() - simulationStartTime,
                        tokenCount);
                tokenBucketCount--;
            } else {
                fprintf(stdout, "\n%012.3Lfms: token t%d arrives,token bucket now has %d tokens",
                        getCurrentTimeMS() - simulationStartTime,
                        tokenCount, tokenBucketCount);
            }
        }
        if (!My402ListEmpty(&queue1)) {
            My402ListElem *firstPacket = My402ListFirst(&queue1);
            if (((struct PacketData *) firstPacket->obj)->numberOfTokens <= tokenBucketCount) {
                tokenBucketCount = tokenBucketCount - ((struct PacketData *) firstPacket->obj)->numberOfTokens;
                //if (My402ListEmpty(&queue2)) {
                struct PacketData *firstPacketForQ2;
                firstPacketForQ2 = ((struct PacketData *) firstPacket->obj);
                My402ListUnlink(&queue1, firstPacket);
                firstPacketForQ2->packetLeavesQ1 = getCurrentTimeMS();
                long double timeInQ1 = firstPacketForQ2->packetLeavesQ1 - firstPacketForQ2->packetEntersQ1;
                statistics.totalTimeInQ1 = statistics.totalTimeInQ1 + timeInQ1;
                fprintf(stdout, "\n%012.3Lfms: p%d leaves q1,time in q1 = %.3Lfms",
                        firstPacketForQ2->packetLeavesQ1 - simulationStartTime,
                        firstPacketForQ2->packetID, timeInQ1);
                if (My402ListEmpty(&queue2)) {
                    My402ListAppend(&queue2, firstPacketForQ2);
                    currentTimeInMilliSeconds = getCurrentTimeMS();
                    My402ListElem *firstPacketInQ2 = My402ListFirst(&queue2);
                    ((struct PacketData *) firstPacketInQ2->obj)->packetEntersQ2 = currentTimeInMilliSeconds;
                    fprintf(stdout, "\n%012.3Lfms: p%d enters q2",
                            ((struct PacketData *) firstPacketInQ2->obj)->packetEntersQ2 - simulationStartTime,
                            ((struct PacketData *) firstPacketInQ2->obj)->packetID);
                    pthread_cond_broadcast(&queueNotEmpty);
                } else {
                    My402ListAppend(&queue2, firstPacketForQ2);
                    currentTimeInMilliSeconds = getCurrentTimeMS();
                    firstPacketForQ2->packetEntersQ2 = currentTimeInMilliSeconds;
                    fprintf(stdout, "\n%012.3Lfms: p%d enters q2",
                            firstPacketForQ2->packetEntersQ2 - simulationStartTime,
                            firstPacketForQ2->packetID);
                }

            }
        }
        tokenThreadServiceTime = getCurrentTimeMS();
        delta = tokenThreadServiceTime - tokenThreadWakeUpTime;
        sleepTime = actualRate - delta;
        pthread_cond_broadcast(&queueNotEmpty);
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

long double getCurrentTimeMS() {
    struct timeval tv;
    gettimeofday(&tv, NULL);
    long double milliSeconds = (tv).tv_sec * 1000.0 + (tv).tv_usec / 1000.0;
    return milliSeconds;
}