-
Notifications
You must be signed in to change notification settings - Fork 29
/
GPI2_SN.c
1654 lines (1391 loc) · 37.5 KB
/
GPI2_SN.c
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
Copyright (c) Fraunhofer ITWM - Carsten Lojewski <[email protected]>, 2013-2016
This file is part of GPI-2.
GPI-2 is free software; you can redistribute it
and/or modify it under the terms of the GNU General Public License
version 3 as published by the Free Software Foundation.
GPI-2 is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with GPI-2. If not, see <http://www.gnu.org/licenses/>.
*/
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/tcp.h>
#include <signal.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/timeb.h>
#include <sys/epoll.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "GPI2.h"
#include "GPI2_CM.h"
#include "GPI2_Dev.h"
#include "GPI2_SN.h"
#include "GPI2_Utility.h"
#include "GPI2_SEG.h"
#define GASPI_EPOLL_CREATE (256)
#define GASPI_EPOLL_MAX_EVENTS (2048)
#define GASPI_SN_RESET_EVENT(mgmt, len, ev) \
mgmt->bdone = 0; \
mgmt->blen = len; \
mgmt->op = ev; \
mgmt->cdh.op = GASPI_SN_RESET;
/* Status and return value of SN thread: mostly for error detection */
volatile enum gaspi_sn_status gaspi_sn_status = GASPI_SN_STATE_INIT;
volatile int _gaspi_sn_stop = 0;
enum
{
GPI2_SN_TIMEOUT = -3,
GPI2_SN_EMFILE = -2,
GPI2_SN_ERROR = -1
};
int
gaspi_sn_set_blocking(const int sock)
{
int flags = fcntl(sock, F_GETFL, 0);
if( flags == -1 )
{
return GPI2_SN_ERROR;
}
flags &= ~O_NONBLOCK;
if( fcntl(sock, F_SETFL, flags) == -1 )
{
return GPI2_SN_ERROR;
}
return 0;
}
int
gaspi_sn_set_non_blocking(const int sock)
{
int sflags = fcntl(sock, F_GETFL, 0);
if( sflags < 0 )
{
return GPI2_SN_ERROR;
}
sflags |= O_NONBLOCK;
if( fcntl(sock, F_SETFL, sflags) < 0 )
{
return GPI2_SN_ERROR;
}
return 0;
}
int
gaspi_sn_set_default_opts(const int sockfd)
{
int opt = 1;
if( setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0 )
{
gaspi_print_error("Failed to set option on socket");
return GPI2_SN_ERROR;
}
if( setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)) < 0 )
{
gaspi_print_error("Failed to set option on socket");
return GPI2_SN_ERROR;
}
return 0;
}
/* check open files limit and try to increase */
static int
_gaspi_check_set_ofile_limit(void)
{
struct rlimit ofiles;
if( getrlimit ( RLIMIT_NOFILE, &ofiles) != 0 )
{
return GPI2_SN_ERROR;
}
if( ofiles.rlim_cur >= ofiles.rlim_max )
{
return GPI2_SN_ERROR;
}
else
{
ofiles.rlim_cur = ofiles.rlim_max;
if( setrlimit(RLIMIT_NOFILE, &ofiles) != 0 )
{
return GPI2_SN_ERROR;
}
}
return 0;
}
static int
gaspi_sn_connect2port_intern(const char * const hn, const unsigned short port)
{
int ret;
int sockfd = -1;
struct sockaddr_in host;
struct hostent* server_data;
sockfd = socket ( AF_INET, SOCK_STREAM, 0 );
if( -1 == sockfd )
{
/* at least deal with open files limit */
int errsv = errno;
if( errsv == EMFILE )
{
if( 0 == _gaspi_check_set_ofile_limit() )
{
sockfd = socket(AF_INET,SOCK_STREAM,0);
if( sockfd == -1 )
{
/* still erroneous */
return GPI2_SN_ERROR;
}
}
else /* failed to check/set ofile limit */
{
return GPI2_SN_EMFILE;
}
}
else
{
return GPI2_SN_ERROR;
}
}
host.sin_family = AF_INET;
host.sin_port = htons(port);
if( (server_data = gethostbyname(hn)) == NULL )
{
close(sockfd);
return GPI2_SN_ERROR;
}
memcpy(&host.sin_addr, server_data->h_addr, server_data->h_length);
/* TODO: we need to be able to distinguish between an initialization
connection attemp and a connection attempt during run-time where
the remote node is gone (FT) */
ret = connect( sockfd, (struct sockaddr *) &host, sizeof(host) );
if( 0 != ret )
{
close( sockfd );
return GPI2_SN_ERROR;
}
if( 0 != gaspi_sn_set_default_opts(sockfd) )
{
gaspi_print_error("Failed to set options on socket.");
close(sockfd);
return GPI2_SN_ERROR;
}
return sockfd;
}
int
gaspi_sn_connect2port(const char * const hn, const unsigned short port, const unsigned long timeout_ms)
{
int sockfd = -1;
struct timeb t0, t1;
const useconds_t max_backoff = 1000000;
useconds_t cur_backoff = 1000;
ftime(&t0);
while( -1 == sockfd )
{
sockfd = gaspi_sn_connect2port_intern(hn, port);
ftime(&t1);
const unsigned int delta_ms = (t1.time - t0.time) * 1000 + (t1.millitm - t0.millitm);
if( sockfd < 0 )
{
if( delta_ms > timeout_ms )
{
return GPI2_SN_TIMEOUT;
}
}
/* exponential backoff */
usleep(cur_backoff);
if( cur_backoff < max_backoff )
{
cur_backoff *= 2;
}
}
signal(SIGPIPE, SIG_IGN);
return sockfd;
}
ssize_t
gaspi_sn_writen(const int sockfd, const void * data_ptr, const size_t n)
{
ssize_t ndone;
size_t left;
char *ptr;
ptr = (char *) data_ptr;
left = n;
while( left > 0 )
{
if( (ndone = write( sockfd, ptr, left) ) <= 0 )
{
if( ndone < 0 && errno == EAGAIN )
{
ndone = 0;
}
else
{
return (-1);
}
}
left -= ndone;
ptr += ndone;
}
return n;
}
int
gaspi_sn_close(const int sockfd)
{
int ret = 0;
if( shutdown(sockfd, SHUT_RDWR) != 0 )
{
ret = 1;
}
if( close(sockfd) != 0 )
{
ret = 1;
}
return ret;
}
ssize_t
gaspi_sn_readn(const int sockfd, const void * data_ptr, const size_t n)
{
ssize_t ndone;
size_t left;
char *ptr;
ptr = (char *) data_ptr;
left = n;
while( left > 0 )
{
/* if( (ndone = read( sockfd, ptr, left) ) <= 0 ) */
/* { */
/* if(ndone < 0 && errno == EAGAIN) */
/* ndone = 0; */
/* else */
/* { */
/* return (-1); */
/* } */
/* } */
if( (ndone = read( sockfd, ptr, left) ) < 0 )
{
if(ndone < 0 && errno == EAGAIN)
ndone = 0;
else
return (-1);
}
else if ( 0 == ndone )
break; /* EOF */
left -= ndone;
ptr += ndone;
}
return (n - left);
}
static int
_gaspi_sn_wait_connection(int port, gaspi_timeout_t timeout_ms)
{
struct sockaddr in_addr;
struct sockaddr_in listeningAddress;
socklen_t in_len = sizeof(in_addr);
int lsock = socket(AF_INET, SOCK_STREAM, 0);
if( lsock < 0 )
{
gaspi_print_error("Failed to create socket.");
return GPI2_SN_ERROR;
}
if( 0 != gaspi_sn_set_default_opts(lsock) )
{
gaspi_print_error("Failed to set socket opts.");
close(lsock);
return GPI2_SN_ERROR;
}
listeningAddress.sin_family = AF_INET;
listeningAddress.sin_port = htons(port);
listeningAddress.sin_addr.s_addr = htonl(INADDR_ANY);
if( bind(lsock, (struct sockaddr*)(&listeningAddress), sizeof(listeningAddress)) < 0 )
{
gaspi_print_error("Failed to bind socket %d", port);
close(lsock);
return GPI2_SN_ERROR;
}
if( listen(lsock, SOMAXCONN) < 0 )
{
gaspi_print_error("Failed to listen on socket");
close(lsock);
return GPI2_SN_ERROR;
}
fd_set rfds;
struct timeval tout;
FD_ZERO (&rfds);
FD_SET (lsock, &rfds);
const long ts = (timeout_ms / 1000);
const long tus = (timeout_ms - ts * 1000) * 1000;
tout.tv_sec = ts;
tout.tv_usec = tus;
const int selret = select (FD_SETSIZE, &rfds, NULL, NULL, &tout);
if( selret < 0 )
{
return GPI2_SN_ERROR;
}
if( selret == 0 )
{
return GPI2_SN_TIMEOUT;
}
int nsock = accept( lsock, &in_addr, &in_len );
if( nsock < 0 )
{
gaspi_print_error("Failed to accept connection.");
close(lsock);
close(nsock);
return GPI2_SN_ERROR;
}
close(lsock);
return nsock;
}
int
gaspi_sn_barrier(const gaspi_timeout_t timeout_ms)
{
gaspi_context_t const * const gctx = &glb_gaspi_ctx;
int rank, src, dst, mask;
int send_val = 1, recv_val = 2;
int size = gctx->tnc;
rank = gctx->rank;
mask = 0x1;
while(mask < size)
{
dst = (rank + mask) % size;
src = (rank - mask + size) % size;
if( gaspi_sn_writen(gctx->sockfd[dst], &send_val, sizeof(send_val)) != sizeof(send_val) )
{
return GPI2_SN_ERROR;
}
if( gaspi_sn_readn(gctx->sockfd[src], &recv_val, sizeof(recv_val)) != sizeof(recv_val) )
{
return GPI2_SN_ERROR;
}
mask <<= 1;
}
return 0;
}
static int
gaspi_sn_recv_topology(gaspi_context_t * const gctx, const gaspi_timeout_t timeout_ms)
{
const int port_to_wait = gctx->config->sn_port + GASPI_MAX_PPN + gctx->localSocket;
int nsock = _gaspi_sn_wait_connection(port_to_wait, timeout_ms);
if( nsock < 0 )
{
return nsock;
}
gaspi_cd_header cdh;
memset(&cdh, 0, sizeof(gaspi_cd_header));
/* Read the header */
if( gaspi_sn_readn(nsock, &cdh, sizeof(cdh)) != sizeof(cdh) )
{
gaspi_print_error("Failed to read topology header.");
close(nsock);
return GPI2_SN_ERROR;
}
gctx->rank = cdh.rank;
gctx->tnc = cdh.tnc;
if( cdh.op != GASPI_SN_TOPOLOGY )
{
gaspi_print_error("Received unexpected topology data.");
}
gctx->hn_poff = (char*) calloc( gctx->tnc, 65 );
if( gctx->hn_poff == NULL )
{
gaspi_print_error("Failed to allocate memory.");
close(nsock);
return GPI2_SN_ERROR;
}
gctx->poff = gctx->hn_poff + gctx->tnc * 64;
/* Read the topology */
if( gaspi_sn_readn(nsock, gctx->hn_poff, gctx->tnc * 65 ) != gctx->tnc * 65 )
{
gaspi_print_error("Failed to read topology data.");
close(nsock);
return GPI2_SN_ERROR;
}
if( gaspi_sn_close( nsock ) != 0 )
{
gaspi_print_error("Failed to close connection.");
return GPI2_SN_ERROR;
}
return 0;
}
static int
gaspi_sn_send_topology(gaspi_context_t * const gctx, const int i, const gaspi_timeout_t timeout_ms)
{
if( (gctx->sockfd[i] = gaspi_sn_connect2port( pgaspi_gethostname(i),
(gctx->config->sn_port + GASPI_MAX_PPN + gctx->poff[i]),
timeout_ms)) < 0 )
{
gaspi_print_error("Failed to connect to %d", i);
return gctx->sockfd[i];
}
if( 0 != gaspi_sn_set_default_opts(gctx->sockfd[i]) )
{
gaspi_print_error("Failed to set socket options");
close(gctx->sockfd[i]);
return GPI2_SN_ERROR;
}
gaspi_cd_header cdh;
memset(&cdh, 0, sizeof(gaspi_cd_header));
cdh.op_len = gctx->tnc * 65; //TODO: 65 is magic
cdh.op = GASPI_SN_TOPOLOGY;
cdh.rank = i;
cdh.tnc = gctx->tnc;
int retval = 0;
ssize_t len = sizeof(gaspi_cd_header);
void* ptr = &cdh;
int sockfd = gctx->sockfd[i];
if( sockfd < 0 )
{
gaspi_print_error("Connection to %d not set", i );
retval = -1;
goto endL;
}
if( gaspi_sn_writen( sockfd, ptr, len) != len )
{
gaspi_print_error("Failed to write topology header to %d.", i);
retval = -1;
goto endL;
}
/* the de facto topology */
ptr = gctx->hn_poff;
len = gctx->tnc * 65;
if( gaspi_sn_writen( sockfd, ptr, len) != len )
{
gaspi_print_error("Failed to write topology data to %d.", i);
retval = -1;
goto endL;
}
endL:
gctx->sockfd[i] = -1;
if( gaspi_sn_close( sockfd ) != 0 )
{
gaspi_print_error("Failed to close connection to %d.", i);
retval = -1;
}
return retval;
}
gaspi_return_t
gaspi_sn_broadcast_topology(gaspi_context_t * const gctx, const gaspi_timeout_t timeout_ms)
{
int mask = 0x1;
int dst, src;
free(gctx->sockfd);
gctx->sockfd = (int *) malloc( gctx->tnc * sizeof(int) );
if( gctx->sockfd == NULL )
{
gaspi_print_error("Failed to allocate memory.");
return GASPI_ERR_MEMALLOC;
}
int i;
for(i = 0; i < gctx->tnc; i++)
{
gctx->sockfd[i] = -1;
}
while( mask <= gctx->tnc )
{
if( gctx->rank & mask )
{
src = gctx->rank - mask;
if( src < 0 )
{
src += gctx->tnc;
}
const int rres = gaspi_sn_recv_topology(gctx, timeout_ms);
if( rres != 0 )
{
if( GPI2_SN_TIMEOUT == rres )
{
return GASPI_TIMEOUT;
}
else
{
return GASPI_ERROR;
}
}
break;
}
mask <<=1;
}
mask >>=1;
while (mask > 0)
{
if( gctx->rank + mask < gctx->tnc)
{
dst = gctx->rank + mask;
if( dst >= gctx->tnc )
{
dst -= gctx->tnc;
}
const int sres = gaspi_sn_send_topology(gctx, dst, timeout_ms);
if( sres != 0 )
{
if( GPI2_SN_TIMEOUT == sres )
{
return GASPI_TIMEOUT;
}
else
{
return GASPI_ERROR;
}
}
}
mask >>=1;
}
return 0;
}
static int
gaspi_sn_segment_register(const gaspi_cd_header snp)
{
gaspi_segment_descriptor_t seg_desc;
seg_desc.rank = snp.rank;
seg_desc.ret = snp.ret;
seg_desc.seg_id = snp.seg_id;
seg_desc.addr = snp.addr;
seg_desc.size = snp.size;
seg_desc.notif_addr = snp.notif_addr;
#ifdef GPI2_CUDA
seg_desc.host_rkey = snp.host_rkey;
seg_desc.host_addr = snp.host_addr;
#endif
#ifdef GPI2_DEVICE_IB
seg_desc.rkey[0] = snp.rkey[0];
seg_desc.rkey[1] = snp.rkey[1];
#endif
return gaspi_segment_set(seg_desc);
}
static gaspi_return_t
gaspi_sn_connect_to_rank(const gaspi_rank_t rank, const gaspi_timeout_t timeout_ms)
{
gaspi_context_t const * const gctx = &glb_gaspi_ctx;
struct timeb t0, t1;
ftime(&t0);
#ifdef DEBUG
if( strcmp(pgaspi_gethostname(rank), "") == 0 )
{
gaspi_print_error("Failed to obtain hostname for rank %u", rank);
return GASPI_ERROR;
}
#endif
/* TODO: introduce backoff delay? */
while(gctx->sockfd[rank] == -1)
{
gctx->sockfd[rank] = gaspi_sn_connect2port(pgaspi_gethostname(rank),
gctx->config->sn_port + gctx->poff[rank],
timeout_ms);
if( -2 == gctx->sockfd[rank] )
{
return GASPI_ERR_EMFILE;
}
if( -1 == gctx->sockfd[rank] )
{
ftime(&t1);
const unsigned int delta_ms = (t1.time - t0.time) * 1000 + (t1.millitm - t0.millitm);
if(delta_ms > timeout_ms)
return GASPI_TIMEOUT;
}
}
return GASPI_SUCCESS;
}
static inline int
gaspi_sn_send_recv_cmd(const gaspi_rank_t target_rank,
enum gaspi_sn_ops op,
void* send_buf, size_t send_size,
void* recv_buf, size_t recv_size)
{
gaspi_context_t const * const gctx = &glb_gaspi_ctx;
const int sockfd = gctx->sockfd[(int) target_rank];
gaspi_cd_header cdh;
memset(&cdh, 0, sizeof(gaspi_cd_header));
cdh.op_len = (int) send_size;
cdh.op = op;
cdh.rank = gctx->rank;
ssize_t wret = gaspi_sn_writen(sockfd, &cdh, sizeof(gaspi_cd_header));
if( wret != sizeof(gaspi_cd_header) )
{
gaspi_print_error("Failed to write to %u", target_rank);
return GPI2_SN_ERROR;
}
wret = gaspi_sn_writen(sockfd, send_buf, send_size);
if( wret != (ssize_t) send_size )
{
gaspi_print_error("Failed to write to %u", target_rank);
return GPI2_SN_ERROR;
}
ssize_t rret = gaspi_sn_readn(sockfd, recv_buf, recv_size);
if( rret != (ssize_t) recv_size )
{
gaspi_print_error("Failed to read from %u", target_rank);
return GPI2_SN_ERROR;
}
return 0;
}
static inline int
_gaspi_sn_connect_command(const gaspi_rank_t rank, void const * const arg)
{
gaspi_dev_exch_info_t* dev_info = (gaspi_dev_exch_info_t* ) arg;
const size_t rc_size = dev_info->info_size;
if( rc_size > 0 )
{
return gaspi_sn_send_recv_cmd(rank, GASPI_SN_CONNECT,
dev_info->local_info, rc_size,
dev_info->remote_info, rc_size);
}
return 0;
}
static inline int
_gaspi_sn_queue_create_command(const gaspi_rank_t rank, const void * const arg)
{
gaspi_dev_exch_info_t* dev_info = (gaspi_dev_exch_info_t* ) arg;
const size_t rc_size = dev_info->info_size;
if( rc_size > 0 )
{
int result = 1;
return gaspi_sn_send_recv_cmd(rank, GASPI_SN_QUEUE_CREATE,
dev_info->local_info, rc_size,
&result, sizeof(int));
}
return 0;
}
static inline int
_gaspi_sn_single_command(const gaspi_rank_t rank, const enum gaspi_sn_ops op)
{
gaspi_context_t const * const gctx = &glb_gaspi_ctx;
gaspi_cd_header cdh;
memset(&cdh, 0, sizeof(gaspi_cd_header));
cdh.op_len = 1;
cdh.op = op;
cdh.rank = rank;
cdh.tnc = gctx->tnc;
ssize_t ret = gaspi_sn_writen(gctx->sockfd[rank], &cdh, sizeof(gaspi_cd_header));
if( ret != sizeof(gaspi_cd_header) )
{
gaspi_print_error("Failed to write to %u (%d %p %lu)",
rank,
gctx->sockfd[rank], &cdh, sizeof(gaspi_cd_header));
return GPI2_SN_ERROR;
}
//TODO: get ack back?
return 0;
}
/*
An allgather operation: each rank in group contributes with its part
(src) of size (size). The result will be in recv buffer (size of
this buffer needs to be size * elements in group.
NOTE that atm NO ordering of data is guaranteed in the recv buffer
ie. that data of rank 0 is in recv[0], rank 1 in recv[1].
*/
int
gaspi_sn_allgather(gaspi_context_t const * const gctx,
void const * const src,
void *const recv, size_t size,
gaspi_group_t group,
gaspi_timeout_t timeout_ms)
{
int left_sock = -1, right_sock = -1;
gaspi_group_ctx_t* grp_ctx = &(gctx->groups[group]);
const int right_rank_in_group = (grp_ctx->rank + grp_ctx->tnc + 1) % grp_ctx->tnc;
const int right_rank = grp_ctx->rank_grp[right_rank_in_group];
const int right_rank_port_offset = gctx->poff[right_rank];
const int my_rank_port_offset = gctx->poff[gctx->rank];
const int port_to_wait = 23333 + my_rank_port_offset;
const int port_to_connect = 23333 + right_rank_port_offset;
/* Connect in a ring */
/* If odd number of ranks, the last rank must connect and then accept */
if( (grp_ctx->rank % 2) == 0 && !( (grp_ctx->rank == grp_ctx->tnc - 1) && (grp_ctx->tnc % 2 != 0) ))
{
left_sock = _gaspi_sn_wait_connection(port_to_wait, timeout_ms);
if( left_sock < 0 )
{
gaspi_print_error("Failed to accept connection on %d(%d).",
port_to_wait, my_rank_port_offset);
return GPI2_SN_ERROR;
}
right_sock = gaspi_sn_connect2port(pgaspi_gethostname( right_rank ),
port_to_connect, timeout_ms);
if( right_sock < 0 )
{
gaspi_print_error("Failed to connect to rank %u on %d (%d).",
right_rank, port_to_connect, right_rank_port_offset);
return GPI2_SN_ERROR;
}
}
else
{
right_sock = gaspi_sn_connect2port(pgaspi_gethostname( right_rank), port_to_connect, timeout_ms);
if( right_sock < 0 )
{
gaspi_print_error("Failed to connect to rank %u on %d (%d).",
right_rank, port_to_connect, right_rank_port_offset);
return GPI2_SN_ERROR;
}
left_sock = _gaspi_sn_wait_connection(port_to_wait, timeout_ms);
if( left_sock < 0 )
{
close(right_sock);
gaspi_print_error("Failed to accept connection on %d(%d).",
port_to_wait, my_rank_port_offset);
return GPI2_SN_ERROR;
}
}
if( 0 != gaspi_sn_set_non_blocking(left_sock) )
{
gaspi_print_error("Failed to set socket");
close(right_sock);
close(left_sock);
return GPI2_SN_ERROR;
}
if( 0 != gaspi_sn_set_non_blocking(right_sock) )
{
gaspi_print_error("Failed to set socket");
close(right_sock);
close(left_sock);
return GPI2_SN_ERROR;
}
ssize_t ret = gaspi_sn_writen(right_sock, src, size);
if( ret != size )
{
gaspi_print_error("Failed to write to %u.", right_rank);
close(right_sock);
close(left_sock);
return GPI2_SN_ERROR;
}
/* copy my part to recv buf */
char* recv_buf = (char*) recv;
memcpy(recv, src, size);
recv_buf += size;
/* exch with peers */
int r;
for(r = 1; r < grp_ctx->tnc; r++)
{
ssize_t rret = gaspi_sn_readn(left_sock, recv_buf, size);
if( rret != size )
{
gaspi_print_error("Failed to read from peer (%u).", grp_ctx->rank_grp[r]);
close(right_sock);
close(left_sock);
return GPI2_SN_ERROR;
}
ret = gaspi_sn_writen(right_sock, recv_buf, size);
if( ret != size )
{
gaspi_print_error("Failed to write to peer (%u).", grp_ctx->rank_grp[r]);
close(right_sock);
close(left_sock);
return GPI2_SN_ERROR;
}
recv_buf += size;
}
shutdown(right_sock, SHUT_WR);
shutdown(left_sock, SHUT_RD);
close(right_sock);
close(left_sock);
return 0;
}
static inline int
_gaspi_sn_segment_register_command(const gaspi_rank_t rank, const void * const arg)
{
gaspi_context_t const * const gctx = &glb_gaspi_ctx;
const gaspi_segment_id_t segment_id = * (gaspi_segment_id_t *) arg;
//TODO: move code to own function (e.g. create_segment_registration_descriptor)
gaspi_cd_header cdh;
memset(&cdh, 0, sizeof(gaspi_cd_header));
cdh.op_len = 0; /* in-place */
cdh.op = GASPI_SN_SEG_REGISTER;
cdh.rank = gctx->rank;
cdh.seg_id = segment_id;
cdh.addr = gctx->rrmd[segment_id][gctx->rank].data.addr;
cdh.notif_addr = gctx->rrmd[segment_id][gctx->rank].notif_spc.addr;
cdh.size = gctx->rrmd[segment_id][gctx->rank].size;
#ifdef GPI2_CUDA
cdh.host_rkey = gctx->rrmd[segment_id][gctx->rank].host_rkey;
cdh.host_addr = gctx->rrmd[segment_id][gctx->rank].host_addr;
#endif
#ifdef GPI2_DEVICE_IB
cdh.rkey[0] = gctx->rrmd[segment_id][gctx->rank].rkey[0];
cdh.rkey[1] = gctx->rrmd[segment_id][gctx->rank].rkey[1];
#endif
ssize_t ret = gaspi_sn_writen(gctx->sockfd[rank], &cdh, sizeof(gaspi_cd_header));
if(ret != sizeof(gaspi_cd_header))
{
gaspi_print_error("Failed to write to rank %u (args: %d %p %lu)",
rank,
gctx->sockfd[rank],
&cdh,
sizeof(gaspi_cd_header));
return GPI2_SN_ERROR;
}
int result = 1;
ssize_t rret = gaspi_sn_readn(gctx->sockfd[rank], &result, sizeof(int));
if( rret != sizeof(int) )
{
gaspi_print_error("Failed to read from rank %u (args: %d %p %lu)",
rank,
gctx->sockfd[rank],
&rret,
sizeof(int));
return GPI2_SN_ERROR;
}
/* Registration failed on the remote side */
if( result != 0 )
{
return GPI2_SN_ERROR;
}