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
133
134
|
#define _DEFAULT_SOURCE
#include <stdint.h>
#include <unistd.h>
#include <string.h>
#include "ipc.h"
#include "main.h"
int send(void * self, local_id dst, const Message * msg) {
if( self == NULL )
return 1;
my_info_t *me = (my_info_t *) self;
if( me->dist_info == NULL )
return 1;
if( dst < 0 || dst > me->dist_info->x )
return 1;
uint8_t base = me->id * (me->dist_info->proccnt - 1); // base of block of pipes for this process
uint8_t idx = 0;
if( me->id < dst )
idx = base + dst - 1;
else
idx = base + dst;
int wr_dscr = me->dist_info->pipes[idx][1];
size_t wr_size = sizeof(msg->s_header) + msg->s_header.s_payload_len;
if( write( wr_dscr, &msg, wr_size ) != wr_size )
return 1;
return 0;
}
int send_multicast(void * self, const Message * msg) {
if( self == NULL )
return 1;
my_info_t *me = (my_info_t *) self;
if( me->dist_info == NULL )
return 1;
size_t wr_size;
int wr_dscr;
uint8_t base = me->id * (me->dist_info->proccnt - 1); // base of block of pipes for this process
uint8_t i;
for( i = 0; i < me->dist_info->proccnt - 1; i++) {
wr_dscr = me->dist_info->pipes[ base + i][1];
wr_size = sizeof(msg->s_header) + msg->s_header.s_payload_len;
if( write( wr_dscr, msg, wr_size ) != wr_size )
return 1;
}
return 0;
}
int receive(void * self, local_id from, Message * msg) {
if( self == NULL )
return 1;
my_info_t *me = (my_info_t *) self;
if( me->dist_info == NULL )
return 1;
if( from < 0 || from > me->dist_info->x )
return 1;
uint8_t base = from * (me->dist_info->proccnt - 1);
uint8_t idx = ( me->id > from ? base + me->id - 1 : base + me->id );
int descriptor = me->dist_info->pipes[idx][0];
char buf[MAX_MESSAGE_LEN];
int size;
while( 1 ) {
if( ( size = read( descriptor, buf, sizeof( buf ) ) ) > 0 ) {
memcpy( msg, buf, sizeof( *msg ) );
return 0;
}
usleep(100000); // sleep 100 ms
}
return -1;
}
int receive_any(void * self, Message * msg) {
if( self == NULL )
return 1;
my_info_t *me = (my_info_t *) self;
if( me->dist_info == NULL )
return 1;
uint8_t i, j = 0;
int descriptors[MAX_PROCCNT - 1];
for(i = 0; i < me->dist_info->proccnt; i++) {
// don't see at pipes as ( ID -> Y )
if( me->id == i )
continue;
// base of block of pipes for this process
uint8_t base = i * (me->dist_info->proccnt - 1);
// index number in block of pipes for ( X -> ID )
uint8_t idx = ( me->id > i ? base + me->id - 1 : base + me->id );
descriptors[j++] = me->dist_info->pipes[idx][0];
}
char buf[MAX_MESSAGE_LEN];
int size;
while( 1 ) {
for( i = 0; i < j; i++ ) {
if( ( size = read( descriptors[i], buf, sizeof( buf ) ) ) > 0 ) {
memcpy( msg, buf, sizeof( *msg ) );
/*switch( msg->s_header.s_type ) {
case STARTED:
fprintf( stderr, "[ID: %d] Message STARTED received\n", me->id );
break;
case DONE:
fprintf( stderr, "[ID: %d] Message DONE received\n", me->id );
break;
}*/
return 0;
}
}
usleep(100000); // sleep 100 ms
}
return 1;
}
|
