network.cc 20.1 KB
Newer Older
1
2
3
4
5
/*!
 *  Copyright (c) 2018 by Contributors
 * \file graph/network.cc
 * \brief DGL networking related APIs
 */
6
#include "./network.h"
7

8
#include <dgl/runtime/container.h>
9
#include <dgl/runtime/ndarray.h>
10
#include <dgl/packed_func_ext.h>
11
12
13
14
15
#include <dgl/immutable_graph.h>
#include <dgl/nodeflow.h>

#include <unordered_map>

16
17
#include "./network/communicator.h"
#include "./network/socket_communicator.h"
18
19
#include "./network/msg_queue.h"
#include "./network/common.h"
20

21
using dgl::network::StringPrintf;
22
using namespace dgl::runtime;
23
24
25
26

namespace dgl {
namespace network {

Chao Ma's avatar
Chao Ma committed
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
NDArray CreateNDArrayFromRaw(std::vector<int64_t> shape,
                             DLDataType dtype,
                             DLContext ctx,
                             void* raw) {
  DLTensor tensor;
  tensor.ctx = ctx;
  tensor.ndim = static_cast<int>(shape.size());
  tensor.dtype = dtype;
  tensor.shape = new int64_t[tensor.ndim];
  for (int i = 0; i < tensor.ndim; ++i) {
    tensor.shape[i] = shape[i];
  }
  tensor.strides = new int64_t[tensor.ndim];
  for (int i = 0; i < tensor.ndim; ++i) {
    tensor.strides[i] = 1;
  }
  for (int i = tensor.ndim - 2; i >= 0; --i) {
    tensor.strides[i] = tensor.shape[i+1] * tensor.strides[i+1];
  }
  tensor.data = raw;
  DLManagedTensor *managed_tensor = new DLManagedTensor();
  managed_tensor->dl_tensor = tensor;
  return NDArray::FromDLPack(managed_tensor);
}

void ArrayMeta::AddArray(const NDArray& array) {
53
54
55
56
57
58
59
60
61
  // We first write the ndim to the data_shape_
  data_shape_.push_back(static_cast<int64_t>(array->ndim));
  // Then we write the data shape
  for (int i = 0; i < array->ndim; ++i) {
    data_shape_.push_back(array->shape[i]);
  }
  ndarray_count_++;
}

Chao Ma's avatar
Chao Ma committed
62
char* ArrayMeta::Serialize(int64_t* size) {
63
64
65
66
67
68
69
70
  char* buffer = nullptr;
  int64_t buffer_size = 0;
  buffer_size += sizeof(msg_type_);
  if (ndarray_count_ != 0) {
    buffer_size += sizeof(ndarray_count_);
    buffer_size += sizeof(data_shape_.size());
    buffer_size += sizeof(int64_t) * data_shape_.size();
  }
71
  // In the future, we should have a better memory management as
Chao Ma's avatar
Chao Ma committed
72
  // allocating a large chunk of memory can be very expensive.
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
  buffer = new char[buffer_size];
  char* pointer = buffer;
  // Write msg_type_
  *(reinterpret_cast<int*>(pointer)) = msg_type_;
  pointer += sizeof(msg_type_);
  if (ndarray_count_ != 0) {
    // Write ndarray_count_
    *(reinterpret_cast<int*>(pointer)) = ndarray_count_;
    pointer += sizeof(ndarray_count_);
    // Write size of data_shape_
    *(reinterpret_cast<size_t*>(pointer)) = data_shape_.size();
    pointer += sizeof(data_shape_.size());
    // Write data of data_shape_
    memcpy(pointer,
        reinterpret_cast<char*>(data_shape_.data()),
        sizeof(int64_t) * data_shape_.size());
89
  }
90
91
  *size = buffer_size;
  return buffer;
92
93
}

Chao Ma's avatar
Chao Ma committed
94
void ArrayMeta::Deserialize(char* buffer, int64_t size) {
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
  int64_t data_size = 0;
  // Read mesg_type_
  msg_type_ = *(reinterpret_cast<int*>(buffer));
  buffer += sizeof(int);
  data_size += sizeof(int);
  if (data_size < size) {
    // Read ndarray_count_
    ndarray_count_ = *(reinterpret_cast<int*>(buffer));
    buffer += sizeof(int);
    data_size += sizeof(int);
    // Read size of data_shape_
    size_t count = *(reinterpret_cast<size_t*>(buffer));
    buffer += sizeof(size_t);
    data_size += sizeof(size_t);
    data_shape_.resize(count);
    // Read data of data_shape_
    memcpy(data_shape_.data(), buffer,
        count * sizeof(int64_t));
    data_size += count * sizeof(int64_t);
114
  }
115
  CHECK_EQ(data_size, size);
116
117
}

Chao Ma's avatar
Chao Ma committed
118
119
120
121
122
123
124
125
126
char* KVStoreMsg::Serialize(int64_t* size) {
  char* buffer = nullptr;
  int64_t buffer_size = 0;
  buffer_size += sizeof(this->msg_type);
  buffer_size += sizeof(this->rank);
  if (!this->name.empty()) {
    buffer_size += sizeof(this->name.size());
    buffer_size += this->name.size();
  }
127
  // In the future, we should have a better memory management as
Chao Ma's avatar
Chao Ma committed
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
  // allocating a large chunk of memory can be very expensive.
  buffer = new char[buffer_size];
  char* pointer = buffer;
  // write msg_type
  *(reinterpret_cast<int*>(pointer)) = this->msg_type;
  pointer += sizeof(this->msg_type);
  // write rank
  *(reinterpret_cast<int*>(pointer)) = this->rank;
  pointer += sizeof(this->rank);
  // write name
  if (!this->name.empty()) {
    *(reinterpret_cast<size_t*>(pointer)) = this->name.size();
    pointer += sizeof(size_t);
    memcpy(pointer, this->name.c_str(), this->name.size());
  }
  *size = buffer_size;
  return buffer;
}

void KVStoreMsg::Deserialize(char* buffer, int64_t size) {
  int64_t data_size = 0;
  // Read msg_type
  this->msg_type = *(reinterpret_cast<int*>(buffer));
  buffer += sizeof(int);
  data_size += sizeof(int);
  // Read rank
  this->rank = *(reinterpret_cast<int*>(buffer));
  buffer += sizeof(int);
  data_size += sizeof(int);
  if (data_size < size) {
    // Read name
    size_t name_size = *(reinterpret_cast<size_t*>(buffer));
    buffer += sizeof(name_size);
    data_size += sizeof(name_size);
    this->name.assign(buffer, name_size);
    data_size += name_size;
  }
  CHECK_EQ(data_size, size);
}

168
169
170
////////////////////////////////// Basic Networking Components ////////////////////////////////


171
172
DGL_REGISTER_GLOBAL("network._CAPI_DGLSenderCreate")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
173
174
175
176
177
178
    std::string type = args[0];
    network::Sender* sender = nullptr;
    if (type == "socket") {
      sender = new network::SocketSender(kQueueSize);
    } else {
      LOG(FATAL) << "Unknown communicator type: " << type;
179
    }
180
    CommunicatorHandle chandle = static_cast<CommunicatorHandle>(sender);
181
182
183
    *rv = chandle;
  });

184
185
186
187
188
189
190
191
192
193
194
195
196
DGL_REGISTER_GLOBAL("network._CAPI_DGLReceiverCreate")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    std::string type = args[0];
    network::Receiver* receiver = nullptr;
    if (type == "socket") {
      receiver = new network::SocketReceiver(kQueueSize);
    } else {
      LOG(FATAL) << "Unknown communicator type: " << type;
    }
    CommunicatorHandle chandle = static_cast<CommunicatorHandle>(receiver);
    *rv = chandle;
  });

197
DGL_REGISTER_GLOBAL("network._CAPI_DGLFinalizeSender")
198
.set_body([] (DGLArgs args, DGLRetValue* rv) {
199
200
201
202
203
    CommunicatorHandle chandle = args[0];
    network::Sender* sender = static_cast<network::Sender*>(chandle);
    sender->Finalize();
  });

204
205
206
207
208
209
210
DGL_REGISTER_GLOBAL("network._CAPI_DGLFinalizeReceiver")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    network::Receiver* receiver = static_cast<network::SocketReceiver*>(chandle);
    receiver->Finalize();
  });

211
212
213
214
215
216
217
DGL_REGISTER_GLOBAL("network._CAPI_DGLSenderAddReceiver")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    std::string ip = args[1];
    int port = args[2];
    int recv_id = args[3];
    network::Sender* sender = static_cast<network::Sender*>(chandle);
218
219
220
221
222
223
224
    std::string addr;
    if (sender->Type() == "socket") {
      addr = StringPrintf("socket://%s:%d", ip.c_str(), port);
    } else {
      LOG(FATAL) << "Unknown communicator type: " << sender->Type();
    }
    sender->AddReceiver(addr.c_str(), recv_id);
225
226
227
228
229
230
231
232
  });

DGL_REGISTER_GLOBAL("network._CAPI_DGLSenderConnect")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    network::Sender* sender = static_cast<network::Sender*>(chandle);
    if (sender->Connect() == false) {
      LOG(FATAL) << "Sender connection failed.";
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
DGL_REGISTER_GLOBAL("network._CAPI_DGLReceiverWait")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    std::string ip = args[1];
    int port = args[2];
    int num_sender = args[3];
    network::Receiver* receiver = static_cast<network::SocketReceiver*>(chandle);
    std::string addr;
    if (receiver->Type() == "socket") {
      addr = StringPrintf("socket://%s:%d", ip.c_str(), port);
    } else {
      LOG(FATAL) << "Unknown communicator type: " << receiver->Type();
    }
    if (receiver->Wait(addr.c_str(), num_sender) == false) {
      LOG(FATAL) << "Wait sender socket failed.";
    }
  });


////////////////////////// Distributed Sampler Components ////////////////////////////////


DGL_REGISTER_GLOBAL("network._CAPI_SenderSendNodeFlow")
259
260
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
261
    int recv_id = args[1];
262
    GraphRef g = args[2];
263
264
265
266
    NDArray node_mapping = args[3];
    NDArray edge_mapping = args[4];
    NDArray layer_offsets = args[5];
    NDArray flow_offsets = args[6];
267
268
    auto ptr = std::dynamic_pointer_cast<ImmutableGraph>(g.sptr());
    CHECK(ptr) << "only immutable graph is allowed in send/recv";
269
    auto csr = ptr->GetInCSR();
270
271
272
273
    // Create a message for the meta data of ndarray
    NDArray indptr = csr->indptr();
    NDArray indice = csr->indices();
    NDArray edge_ids = csr->edge_ids();
Chao Ma's avatar
Chao Ma committed
274
275
276
277
278
279
280
281
    ArrayMeta meta(kNodeFlowMsg);
    meta.AddArray(node_mapping);
    meta.AddArray(edge_mapping);
    meta.AddArray(layer_offsets);
    meta.AddArray(flow_offsets);
    meta.AddArray(indptr);
    meta.AddArray(indice);
    meta.AddArray(edge_ids);
282
283
    // send meta message
    int64_t size = 0;
Chao Ma's avatar
Chao Ma committed
284
    char* data = meta.Serialize(&size);
285
286
287
288
289
    network::Sender* sender = static_cast<network::Sender*>(chandle);
    Message send_msg;
    send_msg.data = data;
    send_msg.size = size;
    send_msg.deallocator = DefaultMessageDeleter;
Chao Ma's avatar
Chao Ma committed
290
    CHECK_EQ(sender->Send(send_msg, recv_id), ADD_SUCCESS);
291
292
293
294
    // send node_mapping
    Message node_mapping_msg;
    node_mapping_msg.data = static_cast<char*>(node_mapping->data);
    node_mapping_msg.size = node_mapping.GetSize();
295
296
    // capture the array in the closure
    node_mapping_msg.deallocator = [node_mapping](Message*) {};
Chao Ma's avatar
Chao Ma committed
297
    CHECK_EQ(sender->Send(node_mapping_msg, recv_id), ADD_SUCCESS);
298
299
300
301
    // send edege_mapping
    Message edge_mapping_msg;
    edge_mapping_msg.data = static_cast<char*>(edge_mapping->data);
    edge_mapping_msg.size = edge_mapping.GetSize();
302
303
    // capture the array in the closure
    edge_mapping_msg.deallocator = [edge_mapping](Message*) {};
Chao Ma's avatar
Chao Ma committed
304
    CHECK_EQ(sender->Send(edge_mapping_msg, recv_id), ADD_SUCCESS);
305
306
307
308
    // send layer_offsets
    Message layer_offsets_msg;
    layer_offsets_msg.data = static_cast<char*>(layer_offsets->data);
    layer_offsets_msg.size = layer_offsets.GetSize();
309
310
    // capture the array in the closure
    layer_offsets_msg.deallocator = [layer_offsets](Message*) {};
Chao Ma's avatar
Chao Ma committed
311
    CHECK_EQ(sender->Send(layer_offsets_msg, recv_id), ADD_SUCCESS);
312
313
314
315
    // send flow_offset
    Message flow_offsets_msg;
    flow_offsets_msg.data = static_cast<char*>(flow_offsets->data);
    flow_offsets_msg.size = flow_offsets.GetSize();
316
317
    // capture the array in the closure
    flow_offsets_msg.deallocator = [flow_offsets](Message*) {};
Chao Ma's avatar
Chao Ma committed
318
    CHECK_EQ(sender->Send(flow_offsets_msg, recv_id), ADD_SUCCESS);
319
320
321
322
    // send csr->indptr
    Message indptr_msg;
    indptr_msg.data = static_cast<char*>(indptr->data);
    indptr_msg.size = indptr.GetSize();
323
324
    // capture the array in the closure
    indptr_msg.deallocator = [indptr](Message*) {};
Chao Ma's avatar
Chao Ma committed
325
    CHECK_EQ(sender->Send(indptr_msg, recv_id), ADD_SUCCESS);
326
327
328
329
    // send csr->indices
    Message indices_msg;
    indices_msg.data = static_cast<char*>(indice->data);
    indices_msg.size = indice.GetSize();
330
331
    // capture the array in the closure
    indices_msg.deallocator = [indice](Message*) {};
Chao Ma's avatar
Chao Ma committed
332
    CHECK_EQ(sender->Send(indices_msg, recv_id), ADD_SUCCESS);
333
334
    // send csr->edge_ids
    Message edge_ids_msg;
335
336
337
338
    edge_ids_msg.data = static_cast<char*>(edge_ids->data);
    edge_ids_msg.size = edge_ids.GetSize();
    // capture the array in the closure
    edge_ids_msg.deallocator = [edge_ids](Message*) {};
Chao Ma's avatar
Chao Ma committed
339
    CHECK_EQ(sender->Send(edge_ids_msg, recv_id), ADD_SUCCESS);
340
341
  });

342
DGL_REGISTER_GLOBAL("network._CAPI_SenderSendSamplerEndSignal")
343
344
345
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    int recv_id = args[1];
Chao Ma's avatar
Chao Ma committed
346
    ArrayMeta meta(kFinalMsg);
347
    int64_t size = 0;
Chao Ma's avatar
Chao Ma committed
348
    char* data = meta.Serialize(&size);
349
    network::Sender* sender = static_cast<network::Sender*>(chandle);
350
351
    Message send_msg = {data, size};
    send_msg.deallocator = DefaultMessageDeleter;
Chao Ma's avatar
Chao Ma committed
352
    CHECK_EQ(sender->Send(send_msg, recv_id), ADD_SUCCESS);
353
354
  });

355
DGL_REGISTER_GLOBAL("network._CAPI_ReceiverRecvNodeFlow")
356
357
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
358
    network::Receiver* receiver = static_cast<network::SocketReceiver*>(chandle);
359
360
    int send_id = 0;
    Message recv_msg;
Chao Ma's avatar
Chao Ma committed
361
362
    CHECK_EQ(receiver->Recv(&recv_msg, &send_id), REMOVE_SUCCESS);
    ArrayMeta meta(recv_msg.data, recv_msg.size);
363
    recv_msg.deallocator(&recv_msg);
Chao Ma's avatar
Chao Ma committed
364
365
    if (meta.msg_type() == kNodeFlowMsg) {
      CHECK_EQ(meta.ndarray_count() * 2, meta.data_shape_.size());
366
      NodeFlow nf = NodeFlow::Create();
367
368
      // node_mapping
      Message array_0;
Chao Ma's avatar
Chao Ma committed
369
370
371
372
373
374
375
      CHECK_EQ(receiver->RecvFrom(&array_0, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[0], 1);
      nf->node_mapping = CreateNDArrayFromRaw(
        {meta.data_shape_[1]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_0.data);
376
377
      // edge_mapping
      Message array_1;
Chao Ma's avatar
Chao Ma committed
378
379
380
381
382
383
384
      CHECK_EQ(receiver->RecvFrom(&array_1, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[2], 1);
      nf->edge_mapping = CreateNDArrayFromRaw(
        {meta.data_shape_[3]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_1.data);
385
386
      // layer_offset
      Message array_2;
Chao Ma's avatar
Chao Ma committed
387
388
389
390
391
392
393
      CHECK_EQ(receiver->RecvFrom(&array_2, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[4], 1);
      nf->layer_offsets = CreateNDArrayFromRaw(
        {meta.data_shape_[5]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_2.data);
394
395
      // flow_offset
      Message array_3;
Chao Ma's avatar
Chao Ma committed
396
397
398
399
400
401
402
      CHECK_EQ(receiver->RecvFrom(&array_3, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[6], 1);
      nf->flow_offsets = CreateNDArrayFromRaw(
        {meta.data_shape_[7]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_3.data);
403
404
      // CSR indptr
      Message array_4;
Chao Ma's avatar
Chao Ma committed
405
406
407
408
409
410
411
      CHECK_EQ(receiver->RecvFrom(&array_4, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[8], 1);
      NDArray indptr = CreateNDArrayFromRaw(
        {meta.data_shape_[9]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_4.data);
412
413
      // CSR indice
      Message array_5;
Chao Ma's avatar
Chao Ma committed
414
415
416
417
418
419
420
      CHECK_EQ(receiver->RecvFrom(&array_5, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[10], 1);
      NDArray indice = CreateNDArrayFromRaw(
        {meta.data_shape_[11]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_5.data);
421
422
      // CSR edge_ids
      Message array_6;
Chao Ma's avatar
Chao Ma committed
423
424
425
426
427
428
429
      CHECK_EQ(receiver->RecvFrom(&array_6, send_id), REMOVE_SUCCESS);
      CHECK_EQ(meta.data_shape_[12], 1);
      NDArray edge_ids = CreateNDArrayFromRaw(
        {meta.data_shape_[13]},
        DLDataType{kDLInt, 64, 1},
        DLContext{kDLCPU, 0},
        array_6.data);
430
431
      // Create CSR
      CSRPtr csr(new CSR(indptr, indice, edge_ids));
432
      nf->graph = GraphPtr(new ImmutableGraph(csr, nullptr));
433
      *rv = nf;
Chao Ma's avatar
Chao Ma committed
434
435
    } else if (meta.msg_type() == kFinalMsg) {
      *rv = meta.msg_type();
436
    } else {
Chao Ma's avatar
Chao Ma committed
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
      LOG(FATAL) << "Unknown message type: " << meta.msg_type();
    }
  });


////////////////////////// Distributed KVStore Components ////////////////////////////////


DGL_REGISTER_GLOBAL("network._CAPI_SenderSendKVMsg")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    int recv_id = args[1];
    KVStoreMsg kv_msg;
    kv_msg.msg_type = args[2];
    kv_msg.rank = args[3];
    network::Sender* sender = static_cast<network::Sender*>(chandle);
    if (kv_msg.msg_type != kFinalMsg && kv_msg.msg_type != kBarrierMsg) {
      std::string name = args[4];
      kv_msg.name = name;
      kv_msg.id = args[5];
      if (kv_msg.msg_type != kPullMsg) {
        kv_msg.data = args[6];
      }
    }
    int64_t kv_size = 0;
    char* kv_data = kv_msg.Serialize(&kv_size);
    // Send kv_data
    Message send_kv_msg;
    send_kv_msg.data = kv_data;
    send_kv_msg.size = kv_size;
    send_kv_msg.deallocator = DefaultMessageDeleter;
    CHECK_EQ(sender->Send(send_kv_msg, recv_id), ADD_SUCCESS);
    if (kv_msg.msg_type != kFinalMsg && kv_msg.msg_type != kBarrierMsg) {
      // Send ArrayMeta
      ArrayMeta meta(kv_msg.msg_type);
      meta.AddArray(kv_msg.id);
      if (kv_msg.msg_type != kPullMsg) {
        meta.AddArray(kv_msg.data);
      }
      int64_t meta_size = 0;
      char* meta_data = meta.Serialize(&meta_size);
      Message send_meta_msg;
      send_meta_msg.data = meta_data;
      send_meta_msg.size = meta_size;
      send_meta_msg.deallocator = DefaultMessageDeleter;
      CHECK_EQ(sender->Send(send_meta_msg, recv_id), ADD_SUCCESS);
      // Send ID NDArray
      Message send_id_msg;
      send_id_msg.data = static_cast<char*>(kv_msg.id->data);
      send_id_msg.size = kv_msg.id.GetSize();
      NDArray id = kv_msg.id;
      send_id_msg.deallocator = [id](Message*) {};
      CHECK_EQ(sender->Send(send_id_msg, recv_id), ADD_SUCCESS);
      // Send data NDArray
      if (kv_msg.msg_type != kPullMsg) {
        Message send_data_msg;
        send_data_msg.data = static_cast<char*>(kv_msg.data->data);
        send_data_msg.size = kv_msg.data.GetSize();
        NDArray data = kv_msg.data;
        send_data_msg.deallocator = [data](Message*) {};
        CHECK_EQ(sender->Send(send_data_msg, recv_id), ADD_SUCCESS);
      }
    }
  });

DGL_REGISTER_GLOBAL("network.CAPI_ReceiverRecvKVMsg")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    CommunicatorHandle chandle = args[0];
    network::Receiver* receiver = static_cast<network::SocketReceiver*>(chandle);
    KVStoreMsg *kv_msg = new KVStoreMsg();
    // Recv kv_Msg
    Message recv_kv_msg;
    int send_id;
    CHECK_EQ(receiver->Recv(&recv_kv_msg, &send_id), REMOVE_SUCCESS);
    kv_msg->Deserialize(recv_kv_msg.data, recv_kv_msg.size);
    recv_kv_msg.deallocator(&recv_kv_msg);
    if (kv_msg->msg_type == kFinalMsg || kv_msg->msg_type == kBarrierMsg) {
      *rv = kv_msg;
      return;
    }
    // Recv ArrayMeta
    Message recv_meta_msg;
    CHECK_EQ(receiver->RecvFrom(&recv_meta_msg, send_id), REMOVE_SUCCESS);
    ArrayMeta meta(recv_meta_msg.data, recv_meta_msg.size);
    recv_meta_msg.deallocator(&recv_meta_msg);
    // Recv ID NDArray
    Message recv_id_msg;
    CHECK_EQ(receiver->RecvFrom(&recv_id_msg, send_id), REMOVE_SUCCESS);
    CHECK_EQ(meta.data_shape_[0], 1);
    kv_msg->id = CreateNDArrayFromRaw(
      {meta.data_shape_[1]},
      DLDataType{kDLInt, 64, 1},
      DLContext{kDLCPU, 0},
      recv_id_msg.data);
    // Recv Data NDArray
    if (kv_msg->msg_type != kPullMsg) {
      Message recv_data_msg;
      CHECK_EQ(receiver->RecvFrom(&recv_data_msg, send_id), REMOVE_SUCCESS);
535
536
537
538
539
      CHECK_GE(meta.data_shape_[2], 1);
      std::vector<int64_t> vec_shape;
      for (int i = 3; i < meta.data_shape_.size(); ++i) {
        vec_shape.push_back(meta.data_shape_[i]);
      }
Chao Ma's avatar
Chao Ma committed
540
      kv_msg->data = CreateNDArrayFromRaw(
541
        vec_shape,
Chao Ma's avatar
Chao Ma committed
542
543
544
        DLDataType{kDLFloat, 32, 1},
        DLContext{kDLCPU, 0},
        recv_data_msg.data);
545
    }
Chao Ma's avatar
Chao Ma committed
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
    *rv = kv_msg;
  });

DGL_REGISTER_GLOBAL("network._CAPI_ReceiverGetKVMsgType")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    KVMsgHandle chandle = args[0];
    network::KVStoreMsg* msg = static_cast<KVStoreMsg*>(chandle);
    *rv = msg->msg_type;
  });

DGL_REGISTER_GLOBAL("network._CAPI_ReceiverGetKVMsgRank")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    KVMsgHandle chandle = args[0];
    network::KVStoreMsg* msg = static_cast<KVStoreMsg*>(chandle);
    *rv = msg->rank;
  });

DGL_REGISTER_GLOBAL("network._CAPI_ReceiverGetKVMsgName")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    KVMsgHandle chandle = args[0];
    network::KVStoreMsg* msg = static_cast<KVStoreMsg*>(chandle);
    *rv = msg->name;
  });

DGL_REGISTER_GLOBAL("network._CAPI_ReceiverGetKVMsgID")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    KVMsgHandle chandle = args[0];
    network::KVStoreMsg* msg = static_cast<KVStoreMsg*>(chandle);
    *rv = msg->id;
  });

DGL_REGISTER_GLOBAL("network._CAPI_ReceiverGetKVMsgData")
.set_body([] (DGLArgs args, DGLRetValue* rv) {
    KVMsgHandle chandle = args[0];
    network::KVStoreMsg* msg = static_cast<KVStoreMsg*>(chandle);
    *rv = msg->data;
582
583
584
585
  });

}  // namespace network
}  // namespace dgl