Merge pull request #52 from talex5/scm-rights · anil.recoil.org/unpac-unpac@299f58f

+14 -10

lib/uring/uring.ml

··· 133 133 (* Used for the sendmsg/recvmsg calls. Liburing doesn't support sendto/recvfrom at the time of writing. *) 134 134 module Msghdr = struct 135 135 type msghdr 136 - type t = msghdr * Sockaddr.t * Iovec.t 137 - external make_msghdr : Sockaddr.t -> Iovec.t-> msghdr = "ocaml_uring_make_msghdr" 136 + type t = msghdr * Sockaddr.t option * Iovec.t 137 + external make_msghdr : int -> Unix.file_descr list -> Sockaddr.t option -> Iovec.t-> msghdr = "ocaml_uring_make_msghdr" 138 + external get_msghdr_fds : msghdr -> Unix.file_descr list = "ocaml_uring_get_msghdr_fds" 138 139 139 - let get_sockaddr (_, addr, _) = addr 140 + let get_fds (hdr, _, _) = get_msghdr_fds hdr 140 141 141 - let create_with_addr addr buffs = 142 + (* Create a value with space for [n_fds] file descriptors. 143 + When sending, [fds] is used to fill those slots. When receiving, they can be left blank. *) 144 + let create_with_addr ~n_fds ~fds ?addr buffs = 142 145 let iovs = Iovec.make buffs in 143 - make_msghdr addr iovs, addr, iovs 146 + make_msghdr n_fds fds addr iovs, addr, iovs 144 147 145 - let create buffs = 146 - create_with_addr (Sockaddr.create ()) buffs 148 + let create ?(n_fds=0) ?addr buffs = 149 + create_with_addr ~n_fds ~fds:[] ?addr buffs 147 150 end 148 151 149 152 type 'a job = 'a Heap.entry ··· 333 336 let accept t fd addr user_data = 334 337 with_id_full t (fun id -> Uring.submit_accept t.uring id fd addr) user_data ~extra_data:addr 335 338 336 - let send_msg t fd addr buffers user_data = 337 - let addr = Sockaddr.of_unix addr in 338 - let msghdr = Msghdr.create_with_addr addr buffers in 339 + let send_msg ?(fds=[]) ?dst t fd buffers user_data = 340 + let addr = Option.map Sockaddr.of_unix dst in 341 + let n_fds = List.length fds in 342 + let msghdr = Msghdr.create_with_addr ~n_fds ~fds ?addr buffers in 339 343 with_id_full t (fun id -> Uring.submit_send_msg t.uring id fd msghdr) user_data ~extra_data:msghdr 340 344 341 345 let recv_msg t fd msghdr user_data =

+12 -11

lib/uring/uring.mli

··· 202 202 module Msghdr : sig 203 203 type t 204 204 205 - val create : Cstruct.t list -> t 205 + val create : ?n_fds:int -> ?addr:Sockaddr.t -> Cstruct.t list -> t 206 206 (** [create buffs] makes a new [msghdr] using the [buffs] 207 - for the underlying [iovec]. A dummy socket address is used 208 - and will be filled when data is received.*) 207 + for the underlying [iovec]. 208 + @param addr The remote address. 209 + Use {!Sockaddr.create} to create a dummy address that will be filled when data is received. 210 + @param n_fds Reserve space to receive this many FDs (default 0) *) 209 211 210 - val get_sockaddr : t -> Sockaddr.t 211 - (** [get_sockaddr t] gets the socket address from [t]. When used 212 - with {!recv_msg} the socket will only be the sender address once the message 213 - is received, until then the address will be a dummy address. *) 212 + val get_fds : t -> Unix.file_descr list 214 213 end 215 214 216 - val send_msg : 'a t -> Unix.file_descr -> Unix.sockaddr -> Cstruct.t list -> 'a -> 'a job option 217 - (** [send_msg t fd addr buffs d] will submit a [sendmsg(2)] request. The [Msghdr] will be constructed 218 - from the address ([addr]) and the buffers ([buffs]). *) 215 + val send_msg : ?fds:Unix.file_descr list -> ?dst:Unix.sockaddr -> 'a t -> Unix.file_descr -> Cstruct.t list -> 'a -> 'a job option 216 + (** [send_msg t fd buffs d] will submit a [sendmsg(2)] request. The [Msghdr] will be constructed 217 + from the FDs ([fds]), address ([dst]) and buffers ([buffs]). 218 + @param dst Destination address. 219 + @param fds Extra file descriptors to attach to the message. *) 219 220 220 221 val recv_msg : 'a t -> Unix.file_descr -> Msghdr.t -> 'a -> 'a job option 221 222 (** [recv_msg t fd msghdr d] will submit a [recvmsg(2)] request. If the request is 222 - successful then the [msghdr] will contain the sender address and the data sent. *) 223 + successful then the [msghdr] will contain the sender address and the data received. *) 223 224 224 225 (** {2 Submitting operations} *) 225 226

+71 -10

lib/uring/uring_stubs.c

··· 27 27 #include <caml/signals.h> 28 28 #include <caml/unixsupport.h> 29 29 #include <caml/socketaddr.h> 30 + #include <sys/socket.h> 30 31 #include <errno.h> 31 32 #include <string.h> 32 33 #include <poll.h> ··· 423 424 custom_fixed_length_default 424 425 }; 425 426 427 + struct msghdr_with_cmsg { 428 + struct msghdr msg; 429 + struct cmsghdr cmsg; 430 + }; 431 + 426 432 // v_sockaddr and v_iov must not be freed before the msghdr as it contains pointers to them 427 433 value 428 - ocaml_uring_make_msghdr(value v_sockaddr, value v_iov) { 429 - CAMLparam2(v_sockaddr, v_iov); 434 + ocaml_uring_make_msghdr(value v_n_fds, value v_fds, value v_sockaddr_opt, value v_iov) { 435 + CAMLparam3(v_fds, v_sockaddr_opt, v_iov); 430 436 CAMLlocal1(v); 431 437 struct msghdr *msg; 432 438 struct iovec *iovs = Iovec_val(Field(v_iov, 0)); 433 439 int iovs_len = Int_val(Field(v_iov, 1)); 440 + int n_fds = Int_val(v_n_fds); 441 + int cmsg_offset, controllen, total_size; 442 + cmsg_offset = sizeof(struct msghdr_with_cmsg) - sizeof(struct cmsghdr); 443 + controllen = n_fds > 0 ? CMSG_SPACE(sizeof(int) * n_fds) : 0; 444 + total_size = cmsg_offset + controllen; 445 + //dprintf("using %d bytes to hold %d FDs\n", total_size, n_fds); 434 446 // Allocate a pointer on the OCaml heap for the msghdr 435 - v = caml_alloc_custom_mem(&msghdr_ops, sizeof(struct msghdr *), sizeof(struct msghdr)); 447 + v = caml_alloc_custom_mem(&msghdr_ops, sizeof(struct msghdr *), total_size); 436 448 Msghdr_val(v) = NULL; 437 - msg = (struct msghdr *) caml_stat_alloc(sizeof(struct msghdr)); 438 - // The msghdr must zero-ed to avoid unwanted errors 439 - memset(msg, 0, sizeof(struct msghdr)); 449 + msg = (struct msghdr *) caml_stat_alloc(total_size); 450 + // The msghdr and cmsghdr must zero-ed to avoid unwanted errors 451 + memset(msg, 0, total_size); 440 452 Msghdr_val(v) = msg; 441 - struct sock_addr_data *addr = Sock_addr_val(v_sockaddr); 442 - // Store the address and iovec data in the message 443 - msg->msg_name = &(addr->sock_addr_addr); 444 - msg->msg_namelen = sizeof(addr->sock_addr_addr); 453 + if (Is_some(v_sockaddr_opt)) { 454 + struct sock_addr_data *addr = Sock_addr_val(Some_val(v_sockaddr_opt)); 455 + // Store the address and iovec data in the message 456 + msg->msg_name = &(addr->sock_addr_addr); 457 + msg->msg_namelen = sizeof(addr->sock_addr_addr); 458 + } else { 459 + msg->msg_name = NULL; 460 + } 445 461 msg->msg_iov = iovs; 446 462 msg->msg_iovlen = iovs_len; 463 + // Add the FDs to the message 464 + if (n_fds > 0) { 465 + int i; 466 + struct cmsghdr *cm; 467 + msg->msg_control = &(((struct msghdr_with_cmsg *) msg)->cmsg); 468 + msg->msg_controllen = controllen; 469 + if (Is_block(v_fds)) { 470 + cm = CMSG_FIRSTHDR(msg); 471 + cm->cmsg_level = SOL_SOCKET; 472 + cm->cmsg_type = SCM_RIGHTS; 473 + cm->cmsg_len = CMSG_LEN(n_fds * sizeof(int)); 474 + for (i = 0; i < n_fds; i++) { 475 + int fd = -1; 476 + if (Is_block(v_fds)) { 477 + fd = Int_val(Field(v_fds, 0)); 478 + v_fds = Field(v_fds, 1); 479 + } 480 + ((int *)CMSG_DATA(cm))[i] = fd; 481 + } 482 + } 483 + } 447 484 CAMLreturn(v); 485 + } 486 + 487 + value 488 + ocaml_uring_get_msghdr_fds(value v_msghdr) { 489 + CAMLparam1(v_msghdr); 490 + CAMLlocal2(v_list, v_cons); 491 + struct msghdr *msg = Msghdr_val(v_msghdr); 492 + struct cmsghdr *cm; 493 + v_list = Val_int(0); 494 + for (cm = CMSG_FIRSTHDR(msg); cm; cm = CMSG_NXTHDR(msg, cm)) { 495 + if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_RIGHTS) { 496 + int *fds = (int *) CMSG_DATA(cm); 497 + int n_fds = (cm->cmsg_len - CMSG_LEN(0)) / sizeof(int); 498 + int i; 499 + for (i = n_fds - 1; i >= 0; i--) { 500 + int fd = Val_int(fds[i]); 501 + value v_cons = caml_alloc_tuple(2); 502 + Store_field(v_cons, 0, fd); 503 + Store_field(v_cons, 1, v_list); 504 + v_list = v_cons; 505 + } 506 + } 507 + } 508 + CAMLreturn(v_list); 448 509 } 449 510 450 511 // v_sockaddr must not be GC'd while the call is in progress

+29

tests/main.ml

··· 366 366 check_int ~__POS__ ~expected:0 r_read; 367 367 Uring.exit t 368 368 369 + let test_send_msg () = 370 + let r, w = Unix.pipe () in 371 + let t = Uring.create ~queue_depth:2 () in 372 + let a, b = Unix.(socketpair PF_UNIX SOCK_STREAM 0) in 373 + let bufs = [Cstruct.of_string "hi"] in 374 + assert_some ~__POS__ (Uring.send_msg t a ~fds:[r; w] bufs `Send); 375 + check_int ~__POS__ (Uring.submit t) ~expected:1; 376 + let _, r_send = consume t in 377 + check_int ~__POS__ ~expected:2 r_send; 378 + let recv_buf = Cstruct.of_string "XX" in 379 + let recv = Uring.Msghdr.create ~n_fds:2 [recv_buf] in 380 + check_int ~__POS__ ~expected:0 (List.length (Uring.Msghdr.get_fds recv)); 381 + assert_some ~__POS__ (Uring.recv_msg t b recv `Recv); 382 + check_int ~__POS__ (Uring.submit t) ~expected:1; 383 + let _, r_recv = consume t in 384 + check_int ~__POS__ ~expected:2 r_recv; 385 + check_string ~__POS__ ~expected:"hi" (Cstruct.to_string recv_buf); 386 + let r2, w2 = 387 + match Uring.Msghdr.get_fds recv with 388 + | [r2; w2] -> r2, w2 389 + | _ -> failwith "Expected two FDs!" 390 + in 391 + check_int ~__POS__ ~expected:5 (Unix.write_substring w2 "to-w2" 0 5); 392 + check_string ~__POS__ ~expected:"to-w2" (really_input_string (Unix.in_channel_of_descr r) 5); 393 + check_int ~__POS__ ~expected:4 (Unix.write_substring w "to-w" 0 4); 394 + check_string ~__POS__ ~expected:"to-w" (really_input_string (Unix.in_channel_of_descr r2) 4); 395 + List.iter Unix.close [r; w; r2; w2] 396 + 369 397 let () = 370 398 Test_data.setup (); 371 399 Random.self_init (); ··· 389 417 tc "cancel" test_cancel; 390 418 tc "cancel_late" test_cancel_late; 391 419 tc "cancel_invalid" test_cancel_invalid; 420 + tc "send_msg" test_send_msg; 392 421 tc "free_busy" test_free_busy; 393 422 ]; 394 423 ]

Configure Feed

Configure Feed