gerrit.fd Code Review - vpp.git/blob - extras/libmemif/src/socket.c

1 /*

2 *------------------------------------------------------------------

4 * Licensed under the Apache License, Version 2.0 (the "License");

5 * you may not use this file except in compliance with the License.

6 * You may obtain a copy of the License at:

7 *

8 * http://www.apache.org/licenses/LICENSE-2.0

9 *

10 * Unless required by applicable law or agreed to in writing, software

11 * distributed under the License is distributed on an "AS IS" BASIS,

12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

13 * See the License for the specific language governing permissions and

14 * limitations under the License.

15 *------------------------------------------------------------------

16 */

18 #define _GNU_SOURCE

19 #include <sys/socket.h>

20 #include <sys/types.h>

21 #include <sys/un.h>

22 #include <string.h>

23 #include <stdio.h>

24 #include <stdlib.h>

25 #include <net/if.h>

26 #include <sys/ioctl.h>

27 #include <sys/uio.h>

28 #include <sys/mman.h>

29 #include <sys/prctl.h>

30 #include <fcntl.h>

31 #include <errno.h>

33 #include <socket.h>

34 #include <memif.h>

35 #include <memif_private.h>

37 /* sends msg to socket */

38 static int

39 memif_msg_send_from_queue (memif_control_channel_t *cc)

40 {

41 struct msghdr mh = { 0 };

42 struct iovec iov[1];

43 char ctl[CMSG_SPACE (sizeof (int))];

44 int rv, err = MEMIF_ERR_SUCCESS; /* 0 */

45 memif_msg_queue_elt_t *e;

47 /* Pick the first message */

48 e = TAILQ_FIRST(&cc->msg_queue);

49 if (e == NULL)

50 return MEMIF_ERR_SUCCESS;

52 /* Construct the message */

53 iov[0].iov_base = (void *) &e->msg;

54 iov[0].iov_len = sizeof (memif_msg_t);

55 mh.msg_iov = iov;

56 mh.msg_iovlen = 1;

58 if (e->fd > 0)

59 {

60 struct cmsghdr *cmsg;

61 memset (&ctl, 0, sizeof (ctl));

62 mh.msg_control = ctl;

63 mh.msg_controllen = sizeof (ctl);

64 cmsg = CMSG_FIRSTHDR (&mh);

65 cmsg->cmsg_len = CMSG_LEN (sizeof (int));

66 cmsg->cmsg_level = SOL_SOCKET;

67 cmsg->cmsg_type = SCM_RIGHTS;

68 memcpy (CMSG_DATA (cmsg), &e->fd, sizeof (int));

69 }

70 rv = sendmsg (cc->fd, &mh, 0);

71 if (rv < 0)

72 err = memif_syscall_error_handler (errno);

73 DBG ("Message type %u sent", e->msg.type);

75 /* If sent successfully, remove the msg from queue */

76 if (err == MEMIF_ERR_SUCCESS) {

77 TAILQ_REMOVE(&cc->msg_queue, e, next);

78 cc->sock->args.free(e);

79 }

81 return err;

82 }

84 static memif_msg_queue_elt_t *

85 memif_msg_enq (memif_control_channel_t *cc)

86 {

87 memif_msg_queue_elt_t *e;

89 e = cc->sock->args.alloc(sizeof(*e));

90 if (e == NULL)

91 return NULL;

93 e->fd = -1;

94 TAILQ_INSERT_TAIL(&cc->msg_queue, e, next);

96 return e;

97 }

99 static int

100 memif_msg_enq_hello (memif_control_channel_t *cc)

101 {

102 memif_msg_hello_t *h;

103 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

104

105 if (e == NULL)

106 return MEMIF_ERR_NOMEM;

107

108 e->msg.type = MEMIF_MSG_TYPE_HELLO;

109

110 h = &e->msg.hello;

111 h->min_version = MEMIF_VERSION;

112 h->max_version = MEMIF_VERSION;

113 h->max_s2m_ring = MEMIF_MAX_S2M_RING;

114 h->max_m2s_ring = MEMIF_MAX_M2S_RING;

115 h->max_region = MEMIF_MAX_REGION;

116 h->max_log2_ring_size = MEMIF_MAX_LOG2_RING_SIZE;

117

118 strncpy ((char *) h->name, (char *) cc->sock->args.app_name,

119 sizeof(h->name));

120

121 return MEMIF_ERR_SUCCESS;

122 }

123

124 /* response from memif master - master is ready to handle next message */

125 static int

126 memif_msg_enq_ack (memif_control_channel_t *cc)

127 {

128 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

129

130 if (e == NULL)

131 return MEMIF_ERR_NOMEM;

132

133 e->msg.type = MEMIF_MSG_TYPE_ACK;

134 e->fd = -1;

135

136 return MEMIF_ERR_SUCCESS; /* 0 */

137 }

138

139 /* send id and secret (optional) for interface identification */

140 static int

141 memif_msg_enq_init (memif_control_channel_t *cc)

142 {

143 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

144

145 if (e == NULL)

146 return MEMIF_ERR_NOMEM;

147

148 memif_msg_init_t *i = &e->msg.init;

149

150 e->msg.type = MEMIF_MSG_TYPE_INIT;

151 e->fd = -1;

152 i->version = MEMIF_VERSION;

153 i->id = cc->conn->args.interface_id;

154 i->mode = cc->conn->args.mode;

155

156 strncpy ((char *) i->name, (char *) cc->sock->args.app_name,

157 sizeof(i->name));

158 if (strlen ((char *) cc->conn->args.secret) > 0)

159 strncpy ((char *) i->secret, (char *) cc->conn->args.secret, sizeof (i->secret));

160

161 return MEMIF_ERR_SUCCESS; /* 0 */

162 }

163

164 /* send information about region specified by region_index */

165 static int

166 memif_msg_enq_add_region (memif_control_channel_t *cc, uint8_t region_index)

167 {

168 memif_region_t *mr = &cc->conn->regions[region_index];

169 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

170

171 if (e == NULL)

172 return MEMIF_ERR_NOMEM;

173

174 memif_msg_add_region_t *ar = &e->msg.add_region;

175

176 e->msg.type = MEMIF_MSG_TYPE_ADD_REGION;

177 e->fd = mr->fd;

178 ar->index = region_index;

179 ar->size = mr->region_size;

180

181 return MEMIF_ERR_SUCCESS; /* 0 */

182 }

183

184 /* send information about ring specified by direction (S2M | M2S) and index */

185 static int

186 memif_msg_enq_add_ring (memif_control_channel_t *cc, uint8_t index, uint8_t dir)

187 {

188 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

189

190 if (e == NULL)

191 return MEMIF_ERR_NOMEM;

192

193 memif_msg_add_ring_t *ar = &e->msg.add_ring;

194

195 e->msg.type = MEMIF_MSG_TYPE_ADD_RING;

196

197 /* TODO: support multiple rings */

198 memif_queue_t *mq;

199 if (dir == MEMIF_RING_M2S)

200 mq = &cc->conn->rx_queues[index];

201 else

202 mq = &cc->conn->tx_queues[index];

203

204 e->fd = mq->int_fd;

205 ar->index = index;

206 ar->offset = mq->offset;

207 ar->region = mq->region;

208 ar->log2_ring_size = mq->log2_ring_size;

209 ar->flags = (dir == MEMIF_RING_S2M) ? MEMIF_MSG_ADD_RING_FLAG_S2M : 0;

210 ar->private_hdr_size = 0;

211

212 return MEMIF_ERR_SUCCESS; /* 0 */

213 }

214

215 /* used as connection request from slave */

216 static int

217 memif_msg_enq_connect (memif_control_channel_t *cc)

218 {

219 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

220

221 if (e == NULL)

222 return MEMIF_ERR_NOMEM;

223

224 memif_msg_connect_t *cm = &e->msg.connect;

225

226 e->msg.type = MEMIF_MSG_TYPE_CONNECT;

227 e->fd = -1;

228 strncpy ((char *) cm->if_name, (char *) cc->conn->args.interface_name,

229 sizeof(cm->if_name));

230

231 return MEMIF_ERR_SUCCESS; /* 0 */

232 }

233

234 /* used as confirmation of connection by master */

235 static int

236 memif_msg_enq_connected (memif_control_channel_t *cc)

237 {

238 memif_msg_queue_elt_t *e = memif_msg_enq(cc);

239

240 if (e == NULL)

241 return MEMIF_ERR_NOMEM;

242

243 memif_msg_connected_t *cm = &e->msg.connected;

244

245 e->msg.type = MEMIF_MSG_TYPE_CONNECTED;

246 e->fd = -1;

247 strncpy ((char *) cm->if_name, (char *) cc->conn->args.interface_name,

248 sizeof(cm->if_name));

249

250 return MEMIF_ERR_SUCCESS; /* 0 */

251 }

252

253 int

254 memif_msg_enq_disconnect (memif_control_channel_t *cc, uint8_t * err_string, uint32_t err_code)

255 {

256 memif_msg_queue_elt_t *e;

257

258 e = cc->sock->args.alloc(sizeof(*e));

259 if (e == NULL)

260 return MEMIF_ERR_NOMEM;

261

262 e->fd = -1;

263 /* Insert disconenct message at the top of the msg queue */

264 TAILQ_INSERT_HEAD(&cc->msg_queue, e, next);

265

266 memif_msg_disconnect_t *d = &e->msg.disconnect;

267

268 e->msg.type = MEMIF_MSG_TYPE_DISCONNECT;

269 d->code = err_code;

270 uint16_t l = sizeof(d->string);

271 if (l > 96)

272 {

273 DBG ("Disconnect string too long. Sending first 96 characters.");

274 l = 96;

275 }

276 strncpy ((char *) d->string, (char *) err_string, l);

277

278 return MEMIF_ERR_SUCCESS;

279 }

280

281 static int

282 memif_msg_parse_hello (memif_control_channel_t *cc, memif_msg_t * msg)

283 {

284 memif_msg_hello_t *h = &msg->hello;

285 memif_connection_t *c = cc->conn;

286

287 if (msg->hello.min_version > MEMIF_VERSION ||

288 msg->hello.max_version < MEMIF_VERSION)

289 {

290 DBG ("incompatible protocol version");

291 return MEMIF_ERR_PROTO;

292 }

293

294 c->run_args.num_s2m_rings = memif_min (h->max_s2m_ring + 1,

295 c->args.num_s2m_rings);

296 c->run_args.num_m2s_rings = memif_min (h->max_m2s_ring + 1,

297 c->args.num_m2s_rings);

298 c->run_args.log2_ring_size = memif_min (h->max_log2_ring_size,

299 c->args.log2_ring_size);

300 c->run_args.buffer_size = c->args.buffer_size;

301 strncpy ((char *) c->remote_name, (char *) h->name,

302 sizeof(c->remote_name));

303

304 return MEMIF_ERR_SUCCESS; /* 0 */

305 }

306

307 /* handle interface identification (id, secret (optional)) */

308 static int

309 memif_msg_parse_init (memif_control_channel_t *cc, memif_msg_t * msg)

310 {

311 memif_msg_init_t *i = &msg->init;

312 memif_connection_t *c = NULL;

313 uint8_t err_string[96];

314 memset (err_string, 0, sizeof (char) * 96);

315 int err = MEMIF_ERR_SUCCESS; /* 0 */

316

317 /* Check compatible meimf version */

318 if (i->version != MEMIF_VERSION)

319 {

320 DBG ("MEMIF_VER_ERR");

321 memif_msg_enq_disconnect(cc, MEMIF_VER_ERR, 0);

322 return MEMIF_ERR_PROTO;

323 }

324

325 /* Find endpoint on the socket */

326 TAILQ_FOREACH(c, &cc->sock->master_interfaces, next)

327 {

328 /* Match interface id */

329 if (c->args.interface_id != i->id)

330 continue;

331 /* If control channel is present, interface is connected (or connecting) */

332 if (c->control_channel != NULL)

333 {

334 memif_msg_enq_disconnect(cc, "Already connected", 0);

335 return MEMIF_ERR_ALRCONN;

336 }

337 /* Verify secret */

338 if (c->args.secret[0] != '\0') {

339 if (strncmp((char *) c->args.secret, (char *) i->secret, 24) != 0)

340 {

341 memif_msg_enq_disconnect(cc, "Incorrect secret", 0);

342 return MEMIF_ERR_SECRET;

343 }

344 }

345

346 /* Assign the control channel to this interface */

347 c->control_channel = cc;

348 cc->conn = c;

349

350 strncpy ((char *) c->remote_name, (char *) i->name,

351 sizeof(c->remote_name));

352 }

353

354 return err;

355 }

356

357 /* receive region information and add new region to connection (if possible) */

358 static int

359 memif_msg_parse_add_region (memif_control_channel_t *cc, memif_msg_t * msg,

360 int fd)

361 {

362 memif_msg_add_region_t *ar = &msg->add_region;

363 memif_region_t *mr;

364 memif_connection_t *c = cc->conn;

365

366 if (fd < 0)

367 return MEMIF_ERR_NO_SHMFD;

368

369 if (ar->index > MEMIF_MAX_REGION)

370 return MEMIF_ERR_MAXREG;

371

372 mr =

373 (memif_region_t *) cc->sock->args.realloc (c->regions,

374 sizeof (memif_region_t) *

375 (++c->regions_num));

376 if (mr == NULL)

377 return memif_syscall_error_handler (errno);

378 memset (mr + ar->index, 0, sizeof (memif_region_t));

379 c->regions = mr;

380 c->regions[ar->index].fd = fd;

381 c->regions[ar->index].region_size = ar->size;

382 c->regions[ar->index].addr = NULL;

383 /* region 0 is never external */

384 if (cc->sock->get_external_region_addr && (ar->index != 0))

385 c->regions[ar->index].is_external = 1;

386

387 return MEMIF_ERR_SUCCESS; /* 0 */

388 }

389

390 /* receive ring information and add new ring to connection queue

391 (based on direction S2M | M2S) */

392 static int

393 memif_msg_parse_add_ring (memif_control_channel_t *cc, memif_msg_t * msg, int fd)

394 {

395 memif_msg_add_ring_t *ar = &msg->add_ring;

396 memif_connection_t *c = cc->conn;

397

398 memif_queue_t *mq;

399

400 if (fd < 0)

401 return MEMIF_ERR_NO_INTFD;

402

403 if (ar->private_hdr_size != 0)

404 return MEMIF_ERR_PRIVHDR;

405

406 if (ar->flags & MEMIF_MSG_ADD_RING_FLAG_S2M)

407 {

408 if (ar->index > MEMIF_MAX_S2M_RING)

409 return MEMIF_ERR_MAXRING;

410 if (ar->index >= c->args.num_s2m_rings)

411 return MEMIF_ERR_MAXRING;

412

413 mq =

414 (memif_queue_t *) cc->sock->args.realloc (c->rx_queues,

415 sizeof (memif_queue_t) *

416 (++c->rx_queues_num));

417 memset (mq + ar->index, 0, sizeof (memif_queue_t));

418 if (mq == NULL)

419 return memif_syscall_error_handler (errno);

420 c->rx_queues = mq;

421 c->rx_queues[ar->index].int_fd = fd;

422 c->rx_queues[ar->index].log2_ring_size = ar->log2_ring_size;

423 c->rx_queues[ar->index].region = ar->region;

424 c->rx_queues[ar->index].offset = ar->offset;

425 c->run_args.num_s2m_rings++;

426 }

427 else

428 {

429 if (ar->index > MEMIF_MAX_M2S_RING)

430 return MEMIF_ERR_MAXRING;

431 if (ar->index >= c->args.num_m2s_rings)

432 return MEMIF_ERR_MAXRING;

433

434 mq =

435 (memif_queue_t *) cc->sock->args.realloc (c->tx_queues,

436 sizeof (memif_queue_t) *

437 (++c->tx_queues_num));

438 memset (mq + ar->index, 0, sizeof (memif_queue_t));

439 if (mq == NULL)

440 return memif_syscall_error_handler (errno);

441 c->tx_queues = mq;

442 c->tx_queues[ar->index].int_fd = fd;

443 c->tx_queues[ar->index].log2_ring_size = ar->log2_ring_size;

444 c->tx_queues[ar->index].region = ar->region;

445 c->tx_queues[ar->index].offset = ar->offset;

446 c->run_args.num_m2s_rings++;

447 }

448

449 return MEMIF_ERR_SUCCESS; /* 0 */

450 }

451

452 static int

453 memif_configure_rx_interrupt (memif_connection_t *c)

454 {

455 memif_socket_t *ms = (memif_socket_t *) c->args.socket;

456 memif_interrupt_t *idata;

457 memif_fd_event_t fde;

458 memif_fd_event_data_t *fdata;

459 void *ctx;

460 int i;

461

462 if (c->on_interrupt != NULL)

463 {

464 for (i = 0; i < c->run_args.num_m2s_rings; i++)

465 {

466 /* Allocate fd event data */

467 fdata = ms->args.alloc(sizeof(*fdata));

468 if (fdata == NULL) {

469 memif_msg_enq_disconnect(c->control_channel, "Internal error", 0);

470 return MEMIF_ERR_NOMEM;

471 }

472 /* Allocate interrupt data */

473 idata = ms->args.alloc(sizeof(*fdata));

474 if (idata == NULL) {

475 ms->args.free(fdata);

476 memif_msg_enq_disconnect(c->control_channel, "Internal error", 0);

477 return MEMIF_ERR_NOMEM;

478 }

479

480 /* configure interrupt data */

481 idata->c = c;

482 idata->qid = i;

483 /* configure fd event data */

484 fdata->event_handler = memif_interrupt_handler;

485 fdata->private_ctx = idata;

486 fde.fd = c->rx_queues[i].int_fd;

487 fde.type = MEMIF_FD_EVENT_READ;

488 fde.private_ctx = fdata;

489

490 /* Start listening for events */

491 ctx = ms->epfd != -1 ? ms : ms->private_ctx;

492 ms->args.on_control_fd_update(fde, ctx);

493 }

494 }

495

496 return MEMIF_ERR_SUCCESS;

497 }

498

499 /* slave -> master */

500 static int

501 memif_msg_parse_connect (memif_control_channel_t *cc, memif_msg_t * msg)

502 {

503 memif_msg_connect_t *cm = &msg->connect;

504 memif_connection_t *c = cc->conn;

505 int err;

506

507 err = memif_connect1 (c);

508 if (err != MEMIF_ERR_SUCCESS)

509 return err;

510

511 strncpy ((char *) c->remote_if_name, (char *) cm->if_name,

512 sizeof(c->remote_if_name));

513

514 err = memif_configure_rx_interrupt(c);

515 if (err != MEMIF_ERR_SUCCESS)

516 return err;

517

518 c->on_connect ((void *) c, c->private_ctx);

519

520 return err;

521 }

522

523 /* master -> slave */

524 static int

525 memif_msg_parse_connected (memif_control_channel_t *cc, memif_msg_t * msg)

526 {

527 memif_msg_connect_t *cm = &msg->connect;

528 memif_connection_t *c = cc->conn;

529

530 int err;

531 err = memif_connect1 (c);

532 if (err != MEMIF_ERR_SUCCESS)

533 return err;

534

535 strncpy ((char *) c->remote_if_name, (char *) cm->if_name,

536 sizeof(c->remote_if_name));

537

538 err = memif_configure_rx_interrupt(c);

539 if (err != MEMIF_ERR_SUCCESS)

540 return err;

541

542 c->on_connect ((void *) c, c->private_ctx);

543

544 return err;

545 }

546

547 static int

548 memif_msg_parse_disconnect (memif_control_channel_t *cc, memif_msg_t * msg)

549 {

550 memif_msg_disconnect_t *d = &msg->disconnect;

551 memif_connection_t *c = cc->conn;

552

553 memset (c->remote_disconnect_string, 0,

554 sizeof (c->remote_disconnect_string));

555 strncpy ((char *) c->remote_disconnect_string, (char *) d->string,

556 sizeof(c->remote_disconnect_string));

557

558 /* on returning error, handle function will call memif_disconnect () */

559 DBG ("disconnect received: %s, mode: %d",

560 c->remote_disconnect_string, c->args.mode);

561 return MEMIF_ERR_DISCONNECT;

562 }

563

564 static int

565 memif_msg_receive_and_parse (memif_control_channel_t *cc)

566 {

567 char ctl[CMSG_SPACE (sizeof (int)) +

568 CMSG_SPACE (sizeof (struct ucred))] = { 0 };

569 struct msghdr mh = { 0 };

570 struct iovec iov[1];

571 memif_msg_t msg = { 0 };

572 ssize_t size;

573 int err = MEMIF_ERR_SUCCESS; /* 0 */

574 int fd = -1;

575 int i;

576 memif_socket_t *ms = NULL;

577

578 iov[0].iov_base = (void *) &msg;

579 iov[0].iov_len = sizeof (memif_msg_t);

580 mh.msg_iov = iov;

581 mh.msg_iovlen = 1;

582 mh.msg_control = ctl;

583 mh.msg_controllen = sizeof (ctl);

584

585 DBG ("recvmsg fd %d", cc->fd);

586 size = recvmsg (cc->fd, &mh, 0);

587 if (size != sizeof (memif_msg_t))

588 {

589 if (size == 0)

590 return MEMIF_ERR_DISCONNECTED;

591 else

592 return MEMIF_ERR_MFMSG;

593 }

594

595 struct cmsghdr *cmsg;

596

597 cmsg = CMSG_FIRSTHDR (&mh);

598 while (cmsg)

599 {

600 if (cmsg->cmsg_level == SOL_SOCKET)

601 {

602 if (cmsg->cmsg_type == SCM_CREDENTIALS)

603 {

604 /* Do nothing */ ;

605 }

606 else if (cmsg->cmsg_type == SCM_RIGHTS)

607 {

608 int *fdp = (int *) CMSG_DATA (cmsg);

609 fd = *fdp;

610 }

611 }

612 cmsg = CMSG_NXTHDR (&mh, cmsg);

613 }

614

615 DBG ("Message type %u received", msg.type);

616

617 switch (msg.type)

618 {

619 case MEMIF_MSG_TYPE_ACK:

620 break;

621

622 case MEMIF_MSG_TYPE_HELLO:

623 if ((err = memif_msg_parse_hello (cc, &msg)) != MEMIF_ERR_SUCCESS)

624 return err;

625 if ((err = memif_init_regions_and_queues (cc->conn)) != MEMIF_ERR_SUCCESS)

626 return err;

627 if ((err = memif_msg_enq_init (cc)) != MEMIF_ERR_SUCCESS)

628 return err;

629 for (i = 0; i < cc->conn->regions_num; i++)

630 {

631 if ((err = memif_msg_enq_add_region (cc, i)) != MEMIF_ERR_SUCCESS)

632 return err;

633 }

634 for (i = 0; i < cc->conn->run_args.num_s2m_rings; i++)

635 {

636 if ((err =

637 memif_msg_enq_add_ring (cc, i,

638 MEMIF_RING_S2M)) != MEMIF_ERR_SUCCESS)

639 return err;

640 }

641 for (i = 0; i < cc->conn->run_args.num_m2s_rings; i++)

642 {

643 if ((err =

644 memif_msg_enq_add_ring (cc, i,

645 MEMIF_RING_M2S)) != MEMIF_ERR_SUCCESS)

646 return err;

647 }

648 if ((err = memif_msg_enq_connect (cc)) != MEMIF_ERR_SUCCESS)

649 return err;

650 break;

651

652 case MEMIF_MSG_TYPE_INIT:

653 if ((err = memif_msg_parse_init (cc, &msg)) != MEMIF_ERR_SUCCESS)

654 return err;

655 /* c->remote_pid = cr->pid */

656 /* c->remote_uid = cr->uid */

657 /* c->remote_gid = cr->gid */

658 if ((err = memif_msg_enq_ack (cc)) != MEMIF_ERR_SUCCESS)

659 return err;

660 break;

661

662 case MEMIF_MSG_TYPE_ADD_REGION:

663 if ((err =

664 memif_msg_parse_add_region (cc, &msg, fd)) != MEMIF_ERR_SUCCESS)

665 return err;

666 if ((err = memif_msg_enq_ack (cc)) != MEMIF_ERR_SUCCESS)

667 return err;

668 break;

669

670 case MEMIF_MSG_TYPE_ADD_RING:

671 if ((err =

672 memif_msg_parse_add_ring (cc, &msg, fd)) != MEMIF_ERR_SUCCESS)

673 return err;

674 if ((err = memif_msg_enq_ack (cc)) != MEMIF_ERR_SUCCESS)

675 return err;

676 break;

677

678 case MEMIF_MSG_TYPE_CONNECT:

679 if ((err = memif_msg_parse_connect (cc, &msg)) != MEMIF_ERR_SUCCESS)

680 return err;

681 if ((err = memif_msg_enq_connected (cc)) != MEMIF_ERR_SUCCESS)

682 return err;

683 break;

684

685 case MEMIF_MSG_TYPE_CONNECTED:

686 if ((err = memif_msg_parse_connected (cc, &msg)) != MEMIF_ERR_SUCCESS)

687 return err;

688 break;

689

690 case MEMIF_MSG_TYPE_DISCONNECT:

691 if ((err = memif_msg_parse_disconnect (cc, &msg)) != MEMIF_ERR_SUCCESS)

692 return err;

693 break;

694

695 default:

696 return MEMIF_ERR_UNKNOWN_MSG;;

697 break;

698 }

699

700 return MEMIF_ERR_SUCCESS; /* 0 */

701 }

702

703 void

704 memif_delete_control_channel (memif_control_channel_t *cc)

705 {

706 memif_msg_queue_elt_t *e, *next;

707 memif_socket_t *ms = cc->sock;

708 memif_fd_event_t fde;

709 void *ctx;

710

711 fde.fd = cc->fd;

712 fde.type = MEMIF_FD_EVENT_DEL;

713 ctx = ms->epfd != -1 ? ms : ms->private_ctx;

714 cc->sock->args.on_control_fd_update(fde, ctx);

715

716 if (cc->fd > 0)

717 close(cc->fd);

718

719 /* Clear control message queue */

720 for (e = TAILQ_FIRST(&cc->msg_queue); e != NULL; e = next) {

721 next = TAILQ_NEXT(e, next);

722 TAILQ_REMOVE(&cc->msg_queue, e, next);

723 cc->sock->args.free(e);

724 }

725

726 /* remove reference */

727 if (cc->conn != NULL)

728 cc->conn->control_channel = NULL;

729 cc->conn = NULL;

730 cc->sock->args.free(cc);

731

732 return;

733 }

734

735

736 int

737 memif_control_channel_handler (memif_fd_event_type_t type, void *private_ctx)

738 {

739 memif_control_channel_t *cc = (memif_control_channel_t *) private_ctx;

740 int err;

741

742 /* Receive the message, parse the message and

743 * enqueue next message(s).

744 */

745 err = memif_msg_receive_and_parse(cc);

746 /* Can't assign to endpoint */

747 if (cc->conn == NULL) {

748 /* A disconnect message is already in the queue */

749 memif_msg_send_from_queue(cc);

750 memif_delete_control_channel(cc);

751

752 return MEMIF_ERR_SUCCESS;

753 }

754 /* error in memif_msg_receive */

755 if (err != MEMIF_ERR_SUCCESS)

756 goto disconnect;

757

758 /* Continue connecting, send next message from the queue */

759 err = memif_msg_send_from_queue(cc);

760 if (err != MEMIF_ERR_SUCCESS)

761 goto disconnect;

762

763 return MEMIF_ERR_SUCCESS;

764

765 disconnect:

766 memif_disconnect_internal(cc->conn);

767 return MEMIF_ERR_SUCCESS;

768 }

769

770 int

771 memif_listener_handler (memif_fd_event_type_t type, void *private_ctx)

772 {

773 memif_socket_t *ms = (memif_socket_t *) private_ctx;

774 memif_control_channel_t *cc;

775 memif_fd_event_t fde;

776 memif_fd_event_data_t *fdata;

777 struct sockaddr_un un;

778 int err, sockfd, addr_len = sizeof(un);

779 void *ctx;

780

781 if (ms == NULL)

782 return MEMIF_ERR_INVAL_ARG;

783

784 if (type & MEMIF_FD_EVENT_READ) {

785 /* Accept connection to the listener socket */

786 sockfd = accept(ms->listener_fd, (struct sockaddr *)&un, (socklen_t *)&addr_len);

787 if (sockfd < 0) {

788 return memif_syscall_error_handler(errno);

789 }

790

791 /* Create new control channel */

792 cc = ms->args.alloc(sizeof(*cc));

793 if (cc == NULL) {

794 err = MEMIF_ERR_NOMEM;

795 goto error;

796 }

797

798 cc->fd = sockfd;

799 /* The connection will be assigned after parsing MEMIF_MSG_TYPE_INIT msg */

800 cc->conn = NULL;

801 cc->sock = ms;

802 TAILQ_INIT(&cc->msg_queue);

803

804 /* Create memif fd event */

805 fdata = ms->args.alloc(sizeof(*fdata));

806 if (fdata == NULL) {

807 err = MEMIF_ERR_NOMEM;

808 goto error;

809 }

810

811 fdata->event_handler = memif_control_channel_handler;

812 fdata->private_ctx = cc;

813

814 fde.fd = sockfd;

815 fde.type = MEMIF_FD_EVENT_READ;

816 fde.private_ctx = fdata;

817

818 /* Start listenning for events on the new control channel */

819 ctx = ms->epfd != -1 ? ms : ms->private_ctx;

820 ms->args.on_control_fd_update(fde, ctx);

821

822 /* enqueue HELLO msg */

823 err = memif_msg_enq_hello(cc);

824 if (err != MEMIF_ERR_SUCCESS)

825 goto error;

826

827 /* send HELLO msg */

828 err = memif_msg_send_from_queue(cc);

829 if (err != MEMIF_ERR_SUCCESS)

830 goto error;

831 }

832

833 return MEMIF_ERR_SUCCESS;

834

835 error:

836 if (sockfd > 0)

837 close(sockfd);

838 if (cc != NULL)

839 ms->args.free(cc);

840 if (fdata != NULL)

841 ms->args.free(fdata);

842

843 return err;

844 }