static int process_response(int wait_for_done, unsigned seq) {
    assert(fd >= 0);
    
    do {
        size_t bytes;
        ssize_t r;
        char replybuf[8*1024];
        char cred_msg[CMSG_SPACE(sizeof(struct ucred))];
        struct msghdr msghdr;
        struct cmsghdr *cmsghdr;
        struct ucred *ucred;
        struct iovec iov; 
        struct nlmsghdr *p = (struct nlmsghdr *) replybuf;
        
        memset(&iov, 0, sizeof(iov));
        iov.iov_base = replybuf; 
 	iov.iov_len = sizeof(replybuf);

        memset(&msghdr, 0, sizeof(msghdr));
        msghdr.msg_name = (void*) NULL;
        msghdr.msg_namelen = 0; 
        msghdr.msg_iov = &iov; 
        msghdr.msg_iovlen = 1; 
        msghdr.msg_control = cred_msg; 
        msghdr.msg_controllen = sizeof(cred_msg); 
 	msghdr.msg_flags = 0;
        
        if ((r = recvmsg(fd, &msghdr, 0)) < 0) {
            daemon_log(LOG_ERR, "recvmsg() failed: %s", strerror(errno));
            return -1;
        }

        if (!(cmsghdr = CMSG_FIRSTHDR(&msghdr)) || cmsghdr->cmsg_type != SCM_CREDENTIALS) {
            daemon_log(LOG_WARNING, "No sender credentials received, ignoring data."); 
            return -1;
        }

        ucred = (struct ucred*) CMSG_DATA(cmsghdr);
        
        if (ucred->uid != 0)
            return -1;

        bytes = (size_t) r;
        
        for (; bytes > 0; p = NLMSG_NEXT(p, bytes)) {

            if (!NLMSG_OK(p, bytes) || bytes < sizeof(struct nlmsghdr) || bytes < p->nlmsg_len) {
                daemon_log(LOG_ERR, "Netlink packet too small.");
                return -1;
            }

            if (p->nlmsg_type == NLMSG_DONE && wait_for_done && p->nlmsg_seq == seq && (pid_t) p->nlmsg_pid == getpid())
                return 0;

            if (p->nlmsg_type == NLMSG_ERROR) {
                struct nlmsgerr *e = (struct nlmsgerr *) NLMSG_DATA (p);

                if (e->error) {
                    daemon_log(LOG_ERR, "Netlink error: %s", strerror(-e->error));
                    return -1;
                }
            }

            if (process_nlmsg(p) < 0)
                return -1;
        }
    } while (wait_for_done);

    return 0;
}

static void routing_packet_read(int fd)
{
    struct in_addr src, dst;
    int i, len, ttl = -1;
    /*AODV_msg *aodv_msg;*/
    struct dev_info *dev;
    struct msghdr msgh;
    struct cmsghdr *cmsg;
    struct iovec iov;
    char ctrlbuf[CMSG_SPACE(sizeof(int)) +
        CMSG_SPACE(sizeof(struct in_pktinfo))];
    struct sockaddr_in src_addr;

    iov.iov_base = recv_buf;
    iov.iov_len = RECV_BUF_SIZE;
    msgh.msg_name = &src_addr;
    msgh.msg_namelen = sizeof(src_addr);
    msgh.msg_iov = &iov;
    msgh.msg_iovlen = 1;
    msgh.msg_control = ctrlbuf;
    msgh.msg_controllen = sizeof(ctrlbuf);

    len = recvmsg(fd, &msgh, 0);

    if (len < 0) {
        fprintf(stderr, "receive ERROR len=%d!", len);
        return;
    }

    src.s_addr = src_addr.sin_addr.s_addr;

    /* Get the ttl and destination address from the control message */
    for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL;
            cmsg = CMSG_NXTHDR_FIX(&msgh, cmsg)) {
        if (cmsg->cmsg_level == SOL_IP) {
            switch (cmsg->cmsg_type) {
                case IP_TTL:
                    ttl = *(CMSG_DATA(cmsg));
                    break;
                case IP_PKTINFO:
                    dst.s_addr =
                        ((struct in_pktinfo *) CMSG_DATA(cmsg))->ipi_addr.s_addr;
            }
        }
    }

    if (ttl < 0) {
        fprintf(stderr, "No TTL, packet ignored!");
        return;
    }

    /* Ignore messages generated locally */
    for (i = 0; i < MAX_NR_INTERFACES; i++)
        if (this_host.devs[i].enabled &&
                memcmp(&src, &this_host.devs[i].ipaddr,
                    sizeof(struct in_addr)) == 0)
            return;

    /*aodv_msg = (AODV_msg *) recv_buf;*/

    /*dev = devfromsock(fd);*/

    callback_set.prot_callback(recv_buf, RECV_BUF_SIZE);

#if 0 
    if (!dev) {
        DEBUG(LOG_ERR, 0, "Could not get device info!\n");
        return;
    }
#endif
}

/*
 * Class:     sun_nio_ch_sctp_SctpChannelImpl
 * Method:    receive0
 * Signature: (ILsun/nio/ch/sctp/ResultContainer;JIZ)I
 */
JNIEXPORT jint JNICALL Java_sun_nio_ch_sctp_SctpChannelImpl_receive0
  (JNIEnv *env, jclass klass, jint fd, jobject resultContainerObj,
   jlong address, jint length, jboolean peek) {
    SOCKADDR sa;
    int sa_len = sizeof(sa);
    ssize_t rv = 0;
    jlong *addr = jlong_to_ptr(address);
    struct iovec iov[1];
    struct msghdr msg[1];
    char cbuf[CMSG_SPACE(sizeof (struct sctp_sndrcvinfo))];
    int flags = peek == JNI_TRUE ? MSG_PEEK : 0;

    /* Set up the msghdr structure for receiving */
    memset(msg, 0, sizeof (*msg));
    msg->msg_name = &sa;
    msg->msg_namelen = sa_len;
    iov->iov_base = addr;
    iov->iov_len = length;
    msg->msg_iov = iov;
    msg->msg_iovlen = 1;
    msg->msg_control = cbuf;
    msg->msg_controllen = sizeof(cbuf);
    msg->msg_flags = 0;

    do {
        if ((rv = recvmsg(fd, msg, flags)) < 0) {
            if (errno == EWOULDBLOCK) {
                return IOS_UNAVAILABLE;
            } else if (errno == EINTR) {
                return IOS_INTERRUPTED;

#ifdef __linux__
            } else if (errno == ENOTCONN) {
                /* ENOTCONN when EOF reached */
                rv = 0;
                /* there will be no control data */
                msg->msg_controllen = 0;
#endif /* __linux__ */

            } else {
                handleSocketError(env, errno);
                return 0;
            }
        }

        if (msg->msg_flags & MSG_NOTIFICATION) {
            char *bufp = (char*)addr;
            union sctp_notification *snp;

            if (!(msg->msg_flags & MSG_EOR) && length < NOTIFICATION_BUFFER_SIZE) {
                char buf[NOTIFICATION_BUFFER_SIZE];
                int rvSAVE = rv;
                memcpy(buf, addr, rv);
                iov->iov_base = buf + rv;
                iov->iov_len = NOTIFICATION_BUFFER_SIZE - rv;
                if ((rv = recvmsg(fd, msg, flags)) < 0) {
                    handleSocketError(env, errno);
                    return 0;
                }
                bufp = buf;
                rv += rvSAVE;
            }
            snp = (union sctp_notification *) bufp;
            if (handleNotification(env, fd, resultContainerObj, snp, rv,
                                   (msg->msg_flags & MSG_EOR),
                                   (struct sockaddr*)&sa ) == JNI_TRUE) {
                /* We have received a notification that is of interest to
                   to the Java API. The appropriate notification will be
                   set in the result container. */
                return 0;
            }

            // set iov back to addr, and reset msg_controllen
            iov->iov_base = addr;
            iov->iov_len = length;
            msg->msg_control = cbuf;
            msg->msg_controllen = sizeof(cbuf);
        }
    } while (msg->msg_flags & MSG_NOTIFICATION);

    handleMessage(env, resultContainerObj, msg, rv,
            (msg->msg_flags & MSG_EOR), (struct sockaddr*)&sa);
    return rv;
}

int main(int argc, char *argv[])
{
        int sk1, sk2;
        sockaddr_storage_t loop1;
        sockaddr_storage_t loop2;
        struct iovec iov;
        struct msghdr inmessage;
	struct msghdr outmessage;
	char incmsg[CMSG_SPACE(sizeof(sctp_cmsg_data_t))];
	char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
	struct cmsghdr *cmsg;
	struct sctp_sndrcvinfo *sinfo;
        struct iovec out_iov;
        int error;
	int pf_class;
	uint32_t ppid;
	uint32_t stream;
	sctp_assoc_t associd1;
	struct sctp_assoc_change *sac;
	struct sctp_event_subscribe subscribe;
	char *big_buffer;
	int offset;
	struct sctp_send_failed *ssf;
	socklen_t len; /* Really becomes 2xlen when set. */
	int orig_len; 
	struct sctp_status gstatus;

        /* Rather than fflush() throughout the code, set stdout to
	 * be unbuffered.
	 */
	setvbuf(stdout, NULL, _IONBF, 0);

	/* Set some basic values which depend on the address family. */
#if TEST_V6
	pf_class = PF_INET6;

        loop1.v6.sin6_family = AF_INET6;
        loop1.v6.sin6_addr = in6addr_loopback;
        loop1.v6.sin6_port = htons(SCTP_TESTPORT_1);

        loop2.v6.sin6_family = AF_INET6;
        loop2.v6.sin6_addr = in6addr_loopback;
        loop2.v6.sin6_port = htons(SCTP_TESTPORT_2);
#else
	pf_class = PF_INET;

        loop1.v4.sin_family = AF_INET;
        loop1.v4.sin_addr.s_addr = SCTP_IP_LOOPBACK;
        loop1.v4.sin_port = htons(SCTP_TESTPORT_1);

        loop2.v4.sin_family = AF_INET;
        loop2.v4.sin_addr.s_addr = SCTP_IP_LOOPBACK;
        loop2.v4.sin_port = htons(SCTP_TESTPORT_2);
#endif /* TEST_V6 */

        /* Create the two endpoints which will talk to each other.  */
        sk1 = test_socket(pf_class, SOCK_SEQPACKET, IPPROTO_SCTP);
        sk2 = test_socket(pf_class, SOCK_SEQPACKET, IPPROTO_SCTP);

	len = sizeof(int);
	error = getsockopt(sk2, SOL_SOCKET, SO_RCVBUF, &orig_len,
			   &len);
	if (error)
		tst_brkm(TBROK, tst_exit, "can't get rcvbuf size: %s",
			strerror(errno));
	/* Set the MAXSEG to something smallish. */
	{
		int val = SMALL_MAXSEG;
		test_setsockopt(sk1, SCTP_MAXSEG, &val, sizeof(val));
	}

	memset(&subscribe, 0, sizeof(subscribe));
	subscribe.sctp_data_io_event = 1;
	subscribe.sctp_association_event = 1;
	subscribe.sctp_send_failure_event = 1;
	test_setsockopt(sk1, SCTP_EVENTS, &subscribe, sizeof(subscribe));
	test_setsockopt(sk2, SCTP_EVENTS, &subscribe, sizeof(subscribe));

        /* Bind these sockets to the test ports.  */
        test_bind(sk1, &loop1.sa, sizeof(loop1));
        test_bind(sk2, &loop2.sa, sizeof(loop2));

	/*
	 * This code sets the associations RWND very small so we can
	 * fill it.  It does this by manipulating the rcvbuf as follows:
	 * 1) Reduce the rcvbuf size on the socket
	 * 2) create an association so that we advertize rcvbuf/2 as
	 *    our initial rwnd
	 * 3) raise the rcvbuf value so that we don't drop data wile 
	 *    receiving later data
	 */
	len = SMALL_RCVBUF;
	error = setsockopt(sk2, SOL_SOCKET, SO_RCVBUF, &len,
			   sizeof(len));
	if (error)
		tst_brkm(TBROK, tst_exit, "setsockopt(SO_RCVBUF): %s",
			 strerror(errno));

       /* Mark sk2 as being able to accept new associations.  */
	test_listen(sk2, 1);
//.........这里部分代码省略.........

static int
read_cmsgspecs(struct thread *td, int ncmsghdrs, struct cmsgspec **cmsgspecs,
	       payload_size_t *payload_size)
{
	struct cmsgspec *spec, *specs;
	payload_size_t actual_payload_size, level_len, nfds_len, type_len;
	size_t datasize, size;
	int error, i, level, nfds, type;

	size = sizeof(specs[0]) * ncmsghdrs;
	specs = (struct cmsgspec *)fmaster_malloc(td, size);
	if (specs == NULL)
		return (ENOMEM);

	actual_payload_size = 0;
	for (i = 0; i < ncmsghdrs; i++) {
		spec = &specs[i];

		error = fmaster_read_int(td, &level, &level_len);
		if (error != 0)
			return (error);
		actual_payload_size += level_len;
		error = fmaster_read_int(td, &type, &type_len);
		if (error != 0)
			return (error);
		actual_payload_size += type_len;
		spec->cmsgspec_level = level;
		spec->cmsgspec_type = type;

		switch (level) {
		case SOL_SOCKET:
			switch (type) {
			case SCM_CREDS:
				datasize = sizeof(struct cmsgcred);
				break;
			case SCM_RIGHTS:
				error = fmaster_read_int(td, &nfds, &nfds_len);
				if (error != 0)
					return (error);
				actual_payload_size += nfds_len;
				spec->cmsgspec_nfds = nfds;
				datasize = sizeof(int) * nfds;
				break;
			default:
				datasize = 0;
				break;
			}
			break;
		default:
			datasize = 0;
			break;
		}

		spec->cmsgspec_len = CMSG_LEN(datasize);
		spec->cmsgspec_space = CMSG_SPACE(datasize);
	}

	*cmsgspecs = specs;
	*payload_size = actual_payload_size;

	return (0);
}

void
ngx_event_recvmsg(ngx_event_t *ev)
{
    ssize_t            n;
    ngx_log_t         *log;
    ngx_err_t          err;
    ngx_event_t       *rev, *wev;
    struct iovec       iov[1];
    struct msghdr      msg;
    ngx_listening_t   *ls;
    ngx_event_conf_t  *ecf;
    ngx_connection_t  *c, *lc;
    u_char             sa[NGX_SOCKADDRLEN];
    static u_char      buffer[65535];

#if (NGX_HAVE_MSGHDR_MSG_CONTROL)

#if (NGX_HAVE_IP_RECVDSTADDR)
    u_char             msg_control[CMSG_SPACE(sizeof(struct in_addr))];
#elif (NGX_HAVE_IP_PKTINFO)
    u_char             msg_control[CMSG_SPACE(sizeof(struct in_pktinfo))];
#endif

#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO)
    u_char             msg_control6[CMSG_SPACE(sizeof(struct in6_pktinfo))];
#endif

#endif

    if (ev->timedout) {
        if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) {
            return;
        }

        ev->timedout = 0;
    }

    ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);

    if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) {
        ev->available = ecf->multi_accept;
    }

    lc = ev->data;
    ls = lc->listening;
    ev->ready = 0;

    ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
                   "recvmsg on %V, ready: %d", &ls->addr_text, ev->available);

    do {
        ngx_memzero(&msg, sizeof(struct msghdr));

        iov[0].iov_base = (void *) buffer;
        iov[0].iov_len = sizeof(buffer);

        msg.msg_name = &sa;
        msg.msg_namelen = sizeof(sa);
        msg.msg_iov = iov;
        msg.msg_iovlen = 1;

#if (NGX_HAVE_MSGHDR_MSG_CONTROL)

        if (ls->wildcard) {

#if (NGX_HAVE_IP_RECVDSTADDR || NGX_HAVE_IP_PKTINFO)
            if (ls->sockaddr->sa_family == AF_INET) {
                msg.msg_control = &msg_control;
                msg.msg_controllen = sizeof(msg_control);
            }
#endif

#if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO)
            if (ls->sockaddr->sa_family == AF_INET6) {
                msg.msg_control = &msg_control6;
                msg.msg_controllen = sizeof(msg_control6);
            }
#endif
        }

#endif

        n = recvmsg(lc->fd, &msg, 0);

        if (n == -1) {
            err = ngx_socket_errno;

            if (err == NGX_EAGAIN) {
                ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, err,
                               "recvmsg() not ready");
                return;
            }

            ngx_log_error(NGX_LOG_ALERT, ev->log, err, "recvmsg() failed");

            return;
        }

#if (NGX_STAT_STUB)
        (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1);
//.........这里部分代码省略.........

  Object* IO::recv_fd(STATE) {
#ifdef _WIN32
    return Primitives::failure();
#else
    struct msghdr msg;
    struct iovec vec[1];
    char buf[1];

    struct cmsghdr *cmsg;
    char cmsg_buf[cmsg_space];

    msg.msg_name = NULL;
    msg.msg_namelen = 0;

    /* Linux and Solaris doesn't work if msg_iov is NULL. */
    buf[0] = '\0';
    vec[0].iov_base = buf;
    vec[0].iov_len = 1;
    msg.msg_iov = vec;
    msg.msg_iovlen = 1;

    msg.msg_control = (caddr_t)cmsg_buf;
    msg.msg_controllen = sizeof(cmsg_buf);
    msg.msg_flags = 0;
    cmsg = CMSG_FIRSTHDR(&msg);
    memset(cmsg_buf, 0, sizeof(cmsg_buf));
    cmsg->cmsg_len = CMSG_LEN(sizeof(int));
    cmsg->cmsg_level = SOL_SOCKET;
    cmsg->cmsg_type = SCM_RIGHTS;

    // Workaround for GCC's broken strict-aliasing checks.
    int* fd_data = (int *)CMSG_DATA(cmsg);
    *fd_data = -1;

    int read_fd = descriptor(state);

    int code = -1;

  retry:
    state->vm()->interrupt_with_signal();
    state->vm()->thread()->sleep(state, cTrue);

    {
      UnmanagedPhase unmanaged(state);
      code = recvmsg(read_fd, &msg, 0);
    }

    state->vm()->thread()->sleep(state, cFalse);
    state->vm()->clear_waiter();

    if(code == -1) {
      if(errno == EAGAIN || errno == EINTR) {
        if(state->vm()->thread_interrupted_p(state)) return NULL;
        ensure_open(state);
        goto retry;
      }

      return Primitives::failure();
    }

    if(msg.msg_controllen != CMSG_SPACE(sizeof(int))
        || cmsg->cmsg_len != CMSG_LEN(sizeof(int))
        || cmsg->cmsg_level != SOL_SOCKET
        || cmsg->cmsg_type != SCM_RIGHTS) {
      return Primitives::failure();
    }

    // Workaround for GCC's broken strict-aliasing checks.
    fd_data = (int *)CMSG_DATA(cmsg);
    return Fixnum::from(*fd_data);
#endif
  }

int J1939SocketRead (int Socket, struct J1939FrameBag *Bags, unsigned int BagsCount, int TimeoutMs)
{
    struct mmsghdr msgs[BagsCount];
    struct iovec iovs[BagsCount];
    struct sockaddr_can addr[BagsCount];
    char ctrlmsgs[BagsCount][
            CMSG_SPACE(sizeof(struct timeval))
          + CMSG_SPACE(sizeof(__u8)) /* dest addr */
          + CMSG_SPACE(sizeof(__u64)) /* dest name */
          + CMSG_SPACE(sizeof(__u8)) /* priority */
          ];

    unsigned int i;
    for (i = 0; i < BagsCount; i++)
    {
        memset(&msgs[i], 0, sizeof(msgs[0]));
        memset(&iovs[i], 0, sizeof(iovs[0]));
        memset(&addr[i], 0, sizeof(addr[0]));

        msgs[i].msg_hdr.msg_name = &addr[i];
        msgs[i].msg_hdr.msg_namelen = sizeof(struct sockaddr_can);

        iovs[i].iov_base = (void *) &(Bags[i].Frame.data);
        iovs[i].iov_len = sizeof(Bags[i].Frame.data);
        msgs[i].msg_hdr.msg_iov = &iovs[i];
        msgs[i].msg_hdr.msg_iovlen = 1;

        msgs[i].msg_hdr.msg_control = &ctrlmsgs[i];
        msgs[i].msg_hdr.msg_controllen = sizeof(ctrlmsgs[0]);
        msgs[i].msg_hdr.msg_flags = 0;
    }

    struct timeval tNow, tEnd;
    for ( initTimers(&tNow, &tEnd, TimeoutMs); timercmp(&tNow, &tEnd, <=); gettimeofday (&tNow, NULL) )
    {
        int rcount;
        rcount = recvmmsg(Socket, msgs, BagsCount, MSG_DONTWAIT, NULL);
        if (rcount >= 0)
        {
            int i;
            for (i = 0; i < rcount; i ++)
            {
                struct timeval tv;
                struct cmsghdr *cmsg;

                for (cmsg = CMSG_FIRSTHDR(&msgs[i].msg_hdr);
                    cmsg;
                    cmsg = CMSG_NXTHDR(&msgs[i].msg_hdr,cmsg))
                {
                    switch (cmsg->cmsg_level) {
                    case SOL_SOCKET:
                        if (cmsg->cmsg_type == SO_TIMESTAMP)
                        {
                            tv = *(struct timeval *)CMSG_DATA(cmsg);
                            Bags[i].TimeStamp.seconds = tv.tv_sec;
                            Bags[i].TimeStamp.microseconds = tv.tv_usec;
                        }
                        else if (cmsg->cmsg_type == SO_RXQ_OVFL)
                            Bags[i].DroppedMessagesCount = *(__u32 *)CMSG_DATA(cmsg);
                        break;
                    case SOL_CAN_J1939:

                        break;
                    }
                }

                Bags[i].Frame.pgn = addr[i].can_addr.j1939.pgn;
                Bags[i].Frame.length = msgs[i].msg_len;

                if (msgs[i].msg_hdr.msg_flags & MSG_CONFIRM)
                    Bags[i].Flags |= (1 << 0);
            }
            return rcount;
        }
        else
        {
            int errsv = errno;
            if (errsv == EAGAIN)
            {
              usleep (100);
              continue;
            }
            else
              return errsv;
        }
    }
    return 0;
}

static PyObject *sendmsg_sendmsg(PyObject *self, PyObject *args, PyObject *keywds) {

    int fd;
    int flags = 0;
    Py_ssize_t sendmsg_result, iovec_length;
    struct msghdr message_header;
    struct iovec iov[1];
    PyObject *ancillary = NULL;
    PyObject *iterator = NULL;
    PyObject *item = NULL;
    PyObject *result_object = NULL;

    static char *kwlist[] = {"fd", "data", "flags", "ancillary", NULL};

    if (!PyArg_ParseTupleAndKeywords(
            args, keywds, "it#|iO:sendmsg", kwlist,
            &fd,
            &iov[0].iov_base,
            &iovec_length,
            &flags,
            &ancillary)) {
        return NULL;
    }

    iov[0].iov_len = iovec_length;

    message_header.msg_name = NULL;
    message_header.msg_namelen = 0;

    message_header.msg_iov = iov;
    message_header.msg_iovlen = 1;

    message_header.msg_control = NULL;
    message_header.msg_controllen = 0;

    message_header.msg_flags = 0;

    if (ancillary) {

        if (!PyList_Check(ancillary)) {
            PyErr_Format(PyExc_TypeError,
                         "send1msg argument 3 expected list, got %s",
                         ancillary->ob_type->tp_name);
            goto finished;
        }

        iterator = PyObject_GetIter(ancillary);

        if (iterator == NULL) {
            goto finished;
        }

        size_t all_data_len = 0;

        /* First we need to know how big the buffer needs to be in order to
           have enough space for all of the messages. */
        while ( (item = PyIter_Next(iterator)) ) {
            int type, level;
            Py_ssize_t data_len;
            size_t prev_all_data_len;
            char *data;

            if (!PyTuple_Check(item)) {
                PyErr_Format(PyExc_TypeError,
                             "send1msg argument 3 expected list of tuple, "
                             "got list containing %s",
                             item->ob_type->tp_name);
                goto finished;
            }

            if (!PyArg_ParseTuple(
                        item, "iit#:sendmsg ancillary data (level, type, data)",
                        &level, &type, &data, &data_len)) {
                goto finished;
            }

            prev_all_data_len = all_data_len;
            all_data_len += CMSG_SPACE(data_len);

            Py_DECREF(item);
            item = NULL;

            if (all_data_len < prev_all_data_len) {
                PyErr_Format(PyExc_OverflowError,
                             "Too much msg_control to fit in a size_t: %zu",
                             prev_all_data_len);
                goto finished;
            }
        }

        Py_DECREF(iterator);
        iterator = NULL;

        /* Allocate the buffer for all of the ancillary elements, if we have
         * any.  */
        if (all_data_len) {
            if (all_data_len > SOCKLEN_MAX) {
                PyErr_Format(PyExc_OverflowError,
                             "Too much msg_control to fit in a socklen_t: %zu",
                             all_data_len);
//.........这里部分代码省略.........

static int dgram_sctp_write(BIO *b, const char *in, int inl)
	{
	int ret;
	bio_dgram_sctp_data *data = (bio_dgram_sctp_data *)b->ptr;
	struct bio_dgram_sctp_sndinfo *sinfo = &(data->sndinfo);
	struct bio_dgram_sctp_prinfo *pinfo = &(data->prinfo);
	struct bio_dgram_sctp_sndinfo handshake_sinfo;
	struct iovec iov[1];
	struct msghdr msg;
	struct cmsghdr *cmsg;
#if defined(SCTP_SNDINFO) && defined(SCTP_PRINFO)
	char cmsgbuf[CMSG_SPACE(sizeof(struct sctp_sndinfo)) + CMSG_SPACE(sizeof(struct sctp_prinfo))];
	struct sctp_sndinfo *sndinfo;
	struct sctp_prinfo *prinfo;
#else
	char cmsgbuf[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
	struct sctp_sndrcvinfo *sndrcvinfo;
#endif

	clear_socket_error();

	/* If we're send anything else than application data,
	 * disable all user parameters and flags.
	 */
	if (in[0] != 23) {
		memset(&handshake_sinfo, 0x00, sizeof(struct bio_dgram_sctp_sndinfo));
#ifdef SCTP_SACK_IMMEDIATELY
		handshake_sinfo.snd_flags = SCTP_SACK_IMMEDIATELY;
#endif
		sinfo = &handshake_sinfo;
	}

	/* If we have to send a shutdown alert message and the
	 * socket is not dry yet, we have to save it and send it
	 * as soon as the socket gets dry.
	 */
	if (data->save_shutdown && !BIO_dgram_sctp_wait_for_dry(b))
	{
		data->saved_message.bio = b;
		data->saved_message.length = inl;
		data->saved_message.data = OPENSSL_malloc(inl);
		memcpy(data->saved_message.data, in, inl);
		return inl;
	}

	iov[0].iov_base = (char *)in;
	iov[0].iov_len = inl;
	msg.msg_name = NULL;
	msg.msg_namelen = 0;
	msg.msg_iov = iov;
	msg.msg_iovlen = 1;
	msg.msg_control = (caddr_t)cmsgbuf;
	msg.msg_controllen = 0;
	msg.msg_flags = 0;
#if defined(SCTP_SNDINFO) && defined(SCTP_PRINFO)
	cmsg = (struct cmsghdr *)cmsgbuf;
	cmsg->cmsg_level = IPPROTO_SCTP;
	cmsg->cmsg_type = SCTP_SNDINFO;
	cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndinfo));
	sndinfo = (struct sctp_sndinfo *)CMSG_DATA(cmsg);
	memset(sndinfo, 0, sizeof(struct sctp_sndinfo));
	sndinfo->snd_sid = sinfo->snd_sid;
	sndinfo->snd_flags = sinfo->snd_flags;
	sndinfo->snd_ppid = sinfo->snd_ppid;
	sndinfo->snd_context = sinfo->snd_context;
	msg.msg_controllen += CMSG_SPACE(sizeof(struct sctp_sndinfo));

	cmsg = (struct cmsghdr *)&cmsgbuf[CMSG_SPACE(sizeof(struct sctp_sndinfo))];
	cmsg->cmsg_level = IPPROTO_SCTP;
	cmsg->cmsg_type = SCTP_PRINFO;
	cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_prinfo));
	prinfo = (struct sctp_prinfo *)CMSG_DATA(cmsg);
	memset(prinfo, 0, sizeof(struct sctp_prinfo));
	prinfo->pr_policy = pinfo->pr_policy;
	prinfo->pr_value = pinfo->pr_value;
	msg.msg_controllen += CMSG_SPACE(sizeof(struct sctp_prinfo));
#else
	cmsg = (struct cmsghdr *)cmsgbuf;
	cmsg->cmsg_level = IPPROTO_SCTP;
	cmsg->cmsg_type = SCTP_SNDRCV;
	cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
	sndrcvinfo = (struct sctp_sndrcvinfo *)CMSG_DATA(cmsg);
	memset(sndrcvinfo, 0, sizeof(struct sctp_sndrcvinfo));
	sndrcvinfo->sinfo_stream = sinfo->snd_sid;
	sndrcvinfo->sinfo_flags = sinfo->snd_flags;
#ifdef __FreeBSD__
	sndrcvinfo->sinfo_flags |= pinfo->pr_policy;
#endif
	sndrcvinfo->sinfo_ppid = sinfo->snd_ppid;
	sndrcvinfo->sinfo_context = sinfo->snd_context;
	sndrcvinfo->sinfo_timetolive = pinfo->pr_value;
	msg.msg_controllen += CMSG_SPACE(sizeof(struct sctp_sndrcvinfo));
#endif

	ret = sendmsg(b->num, &msg, 0);

	BIO_clear_retry_flags(b);
	if (ret <= 0)
		{
		if (BIO_dgram_should_retry(ret))
//.........这里部分代码省略.........

/**
 * Send netlink message.
 * @arg sk		Netlink socket.
 * @arg msg		Netlink message to be sent.
 * @arg iov		iovec to be sent.
 * @arg iovlen		number of struct iovec to be sent.
 * @see nl_sendmsg()
 * @return Number of characters sent on success or a negative error code.
 */
int nl_send_iovec(struct nl_sock *sk, struct nl_msg *msg, struct iovec *iov, unsigned iovlen)
{
	struct sockaddr_nl *dst;
	struct ucred *creds;
	struct msghdr hdr = {
		.msg_name = (void *) &sk->s_peer,
		.msg_namelen = sizeof(struct sockaddr_nl),
		.msg_iov = iov,
		.msg_iovlen = iovlen,
	};

	/* Overwrite destination if specified in the message itself, defaults
	 * to the peer address of the socket.
	 */
	dst = nlmsg_get_dst(msg);
	if (dst->nl_family == AF_NETLINK)
		hdr.msg_name = dst;

	/* Add credentials if present. */
	creds = nlmsg_get_creds(msg);
	if (creds != NULL) {
		char buf[CMSG_SPACE(sizeof(struct ucred))];
		struct cmsghdr *cmsg;

		hdr.msg_control = buf;
		hdr.msg_controllen = sizeof(buf);

		cmsg = CMSG_FIRSTHDR(&hdr);
		cmsg->cmsg_level = SOL_SOCKET;
		cmsg->cmsg_type = SCM_CREDENTIALS;
		cmsg->cmsg_len = CMSG_LEN(sizeof(struct ucred));
		memcpy(CMSG_DATA(cmsg), creds, sizeof(struct ucred));
	}

	return nl_sendmsg(sk, msg, &hdr);
}



/**
* Send netlink message.
* @arg sk		Netlink socket.
* @arg msg		Netlink message to be sent.
* @see nl_sendmsg()
* @return Number of characters sent on success or a negative error code.
*/
int nl_send(struct nl_sock *sk, struct nl_msg *msg)
{
	struct iovec iov = {
		.iov_base = (void *) nlmsg_hdr(msg),
		.iov_len = nlmsg_hdr(msg)->nlmsg_len,
	};

	return nl_send_iovec(sk, msg, &iov, 1);
}

void nl_auto_complete(struct nl_sock *sk, struct nl_msg *msg)
{
	struct nlmsghdr *nlh;

	nlh = nlmsg_hdr(msg);
	if (nlh->nlmsg_pid == 0)
		nlh->nlmsg_pid = sk->s_local.nl_pid;

	if (nlh->nlmsg_seq == 0)
		nlh->nlmsg_seq = sk->s_seq_next++;

	if (msg->nm_protocol == -1)
		msg->nm_protocol = sk->s_proto;

	nlh->nlmsg_flags |= NLM_F_REQUEST;

	if (!(sk->s_flags & NL_NO_AUTO_ACK))
		nlh->nlmsg_flags |= NLM_F_ACK;
}

/**
 * Send netlink message and check & extend header values as needed.
 * @arg sk		Netlink socket.
 * @arg msg		Netlink message to be sent.
 *
 * Checks the netlink message \c nlh for completness and extends it
 * as required before sending it out. Checked fields include pid,
 * sequence nr, and flags.
 *
 * @see nl_send()
 * @return Number of characters sent or a negative error code.
 */
int nl_send_auto_complete(struct nl_sock *sk, struct nl_msg *msg)
{
	struct nl_cb *cb = sk->s_cb;
//.........这里部分代码省略.........

bool Connection::sendOutgoingMessage(std::unique_ptr<MessageEncoder> encoder)
{
    COMPILE_ASSERT(sizeof(MessageInfo) + attachmentMaxAmount * sizeof(size_t) <= messageMaxSize, AttachmentsFitToMessageInline);

    Vector<Attachment> attachments = encoder->releaseAttachments();
    AttachmentResourceGuard<Vector<Attachment>, Vector<Attachment>::iterator> attachementDisposer(attachments);

    if (attachments.size() > (attachmentMaxAmount - 1)) {
        ASSERT_NOT_REACHED();
        return false;
    }

    MessageInfo messageInfo(encoder->bufferSize(), attachments.size());
    size_t messageSizeWithBodyInline = sizeof(messageInfo) + (attachments.size() * sizeof(AttachmentInfo)) + encoder->bufferSize();
    if (messageSizeWithBodyInline > messageMaxSize && encoder->bufferSize()) {
        RefPtr<WebKit::SharedMemory> oolMessageBody = WebKit::SharedMemory::create(encoder->bufferSize());
        if (!oolMessageBody)
            return false;

        WebKit::SharedMemory::Handle handle;
        if (!oolMessageBody->createHandle(handle, WebKit::SharedMemory::ReadOnly))
            return false;

        messageInfo.setMessageBodyIsOutOfLine();

        memcpy(oolMessageBody->data(), encoder->buffer(), encoder->bufferSize());

        attachments.append(handle.releaseToAttachment());
    }

    struct msghdr message;
    memset(&message, 0, sizeof(message));

    struct iovec iov[3];
    memset(&iov, 0, sizeof(iov));

    message.msg_iov = iov;
    int iovLength = 1;

    iov[0].iov_base = reinterpret_cast<void*>(&messageInfo);
    iov[0].iov_len = sizeof(messageInfo);

    auto attachmentInfo = std::make_unique<AttachmentInfo[]>(attachments.size());

    size_t attachmentFDBufferLength = 0;
    if (!attachments.isEmpty()) {
        for (size_t i = 0; i < attachments.size(); ++i) {
            if (attachments[i].fileDescriptor() != -1)
                attachmentFDBufferLength++;
        }
    }
    auto attachmentFDBuffer = std::make_unique<char[]>(CMSG_SPACE(sizeof(int) * attachmentFDBufferLength));

    if (!attachments.isEmpty()) {
        int* fdPtr = 0;

        if (attachmentFDBufferLength) {
            message.msg_control = attachmentFDBuffer.get();
            message.msg_controllen = CMSG_SPACE(sizeof(int) * attachmentFDBufferLength);
            memset(message.msg_control, 0, message.msg_controllen);

            struct cmsghdr* cmsg = CMSG_FIRSTHDR(&message);
            cmsg->cmsg_level = SOL_SOCKET;
            cmsg->cmsg_type = SCM_RIGHTS;
            cmsg->cmsg_len = CMSG_LEN(sizeof(int) * attachmentFDBufferLength);

            fdPtr = reinterpret_cast<int*>(CMSG_DATA(cmsg));
        }

        int fdIndex = 0;
        for (size_t i = 0; i < attachments.size(); ++i) {
            attachmentInfo[i].setType(attachments[i].type());

            switch (attachments[i].type()) {
            case Attachment::MappedMemoryType:
                attachmentInfo[i].setSize(attachments[i].size());
                // Fall trhough, set file descriptor or null.
            case Attachment::SocketType:
                if (attachments[i].fileDescriptor() != -1) {
                    ASSERT(fdPtr);
                    fdPtr[fdIndex++] = attachments[i].fileDescriptor();
                } else
                    attachmentInfo[i].setNull();
                break;
            case Attachment::Uninitialized:
            default:
                break;
            }
        }

        iov[iovLength].iov_base = attachmentInfo.get();
        iov[iovLength].iov_len = sizeof(AttachmentInfo) * attachments.size();
        ++iovLength;
    }

    if (!messageInfo.isMessageBodyIsOutOfLine() && encoder->bufferSize()) {
        iov[iovLength].iov_base = reinterpret_cast<void*>(encoder->buffer());
        iov[iovLength].iov_len = encoder->bufferSize();
        ++iovLength;
    }
//.........这里部分代码省略.........

/**
 * Install the seccomp-bpf that generates trace traps for all syscalls
 * other than those made through _untraced_syscall_entry_point().
 */
static void install_syscall_filter(void)
{
	void* untraced_syscall_start = get_untraced_syscall_entry_point();
	struct sock_filter filter[] = {
		/* Allow all system calls from our protected_call
		 * callsite */
		ALLOW_SYSCALLS_FROM_CALLSITE((uintptr_t)untraced_syscall_start),
		/* All the rest are handled in rr */
		TRACE_PROCESS,
	};
	struct sock_fprog prog = {
		.len = (unsigned short)(sizeof(filter)/sizeof(filter[0])),
		.filter = filter,
	};

	debug("Initializing syscall buffer: protected_call_start = %p",
	      untraced_syscall_start);

	if (traced_prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0)) {
		fatal("prctl(NO_NEW_PRIVS) failed, SECCOMP_FILTER is not available: your kernel is too old.  Use `record -n` to disable the filter.");
	}

	/* Note: the filter is installed only for record. This call
	 * will be emulated in the replay */
	if (traced_prctl(PR_SET_SECCOMP, SECCOMP_MODE_FILTER,
			 (uintptr_t)&prog, 0, 0)) {
		fatal("prctl(SECCOMP) failed, SECCOMP_FILTER is not available: your kernel is too old.  Use `record -n` to disable the filter.");
	}
	/* anything that happens from this point on gets filtered! */
}

/**
 * Return a counter that generates a signal targeted at this task
 * every time the task is descheduled |nr_descheds| times.
 */
static int open_desched_event_counter(size_t nr_descheds, pid_t tid)
{
	struct perf_event_attr attr;
	int fd;
	struct f_owner_ex own;

	memset(&attr, 0, sizeof(attr));
	attr.size = sizeof(attr);
	attr.type = PERF_TYPE_SOFTWARE;
	attr.config = PERF_COUNT_SW_CONTEXT_SWITCHES;
	attr.disabled = 1;
	attr.sample_period = nr_descheds;

	fd = traced_perf_event_open(&attr, 0/*self*/, -1/*any cpu*/, -1, 0);
	if (0 > fd) {
		fatal("Failed to perf_event_open(cs, period=%u)", nr_descheds);
	}
	if (traced_fcntl(fd, F_SETFL, O_ASYNC)) {
		fatal("Failed to fcntl(O_ASYNC) the desched counter");
	}
	own.type = F_OWNER_TID;
	own.pid = tid;
	if (traced_fcntl(fd, F_SETOWN_EX, &own)) {
		fatal("Failed to fcntl(SETOWN_EX) the desched counter to this");
	}
	if (traced_fcntl(fd, F_SETSIG, SYSCALLBUF_DESCHED_SIGNAL)) {
		fatal("Failed to fcntl(SETSIG, %d) the desched counter",
		      SYSCALLBUF_DESCHED_SIGNAL);
	}

	return fd;
}

static void set_up_buffer(void)
{
	struct sockaddr_un addr;
	struct msghdr msg;
	struct iovec data;
	int msgbuf;
	struct cmsghdr* cmsg;
	int* msg_fdptr;
	int* cmsg_fdptr;
	char cmsgbuf[CMSG_SPACE(sizeof(*cmsg_fdptr))];
	struct socketcall_args args_vec;
	struct rrcall_init_buffers_params args;
	pid_t tid = traced_gettid();

	assert(!buffer);

	/* NB: we want this setup emulated during replay. */
	if (buffer_enabled) {
		desched_counter_fd = open_desched_event_counter(1, tid);
	}

	/* Prepare arguments for rrcall.  We do this in the tracee
	 * just to avoid some hairy IPC to set up the arguments
	 * remotely from the tracer; this isn't strictly
	 * necessary. */
	prepare_syscallbuf_socket_addr(&addr, tid);

	memset(&msg, 0, sizeof(msg));
//.........这里部分代码省略.........

static PyObject *sendmsg_recvmsg(PyObject *self, PyObject *args, PyObject *keywds) {
    int fd = -1;
    int flags = 0;
    int maxsize = 8192;
    int cmsg_size = 4096;
    size_t cmsg_space;
    size_t cmsg_overhead;
    Py_ssize_t recvmsg_result;

    struct msghdr message_header;
    struct cmsghdr *control_message;
    struct iovec iov[1];
    char *cmsgbuf;
    PyObject *ancillary;
    PyObject *final_result = NULL;

    static char *kwlist[] = {"fd", "flags", "maxsize", "cmsg_size", NULL};

    if (!PyArg_ParseTupleAndKeywords(args, keywds, "i|iii:recvmsg", kwlist,
                                     &fd, &flags, &maxsize, &cmsg_size)) {
        return NULL;
    }

    cmsg_space = CMSG_SPACE(cmsg_size);

    /* overflow check */
    if (cmsg_space > SOCKLEN_MAX) {
        PyErr_Format(PyExc_OverflowError,
                     "CMSG_SPACE(cmsg_size) greater than SOCKLEN_MAX: %d",
                     cmsg_size);
        return NULL;
    }

    message_header.msg_name = NULL;
    message_header.msg_namelen = 0;

    iov[0].iov_len = maxsize;
    iov[0].iov_base = PyMem_Malloc(maxsize);

    if (!iov[0].iov_base) {
        PyErr_NoMemory();
        return NULL;
    }

    message_header.msg_iov = iov;
    message_header.msg_iovlen = 1;

    cmsgbuf = PyMem_Malloc(cmsg_space);

    if (!cmsgbuf) {
        PyMem_Free(iov[0].iov_base);
        PyErr_NoMemory();
        return NULL;
    }

    memset(cmsgbuf, 0, cmsg_space);
    message_header.msg_control = cmsgbuf;
    /* see above for overflow check */
    message_header.msg_controllen = (socklen_t) cmsg_space;

    recvmsg_result = recvmsg(fd, &message_header, flags);
    if (recvmsg_result < 0) {
        PyErr_SetFromErrno(sendmsg_socket_error);
        goto finished;
    }

    ancillary = PyList_New(0);
    if (!ancillary) {
        goto finished;
    }

    for (control_message = CMSG_FIRSTHDR(&message_header);
         control_message;
         control_message = CMSG_NXTHDR(&message_header,
                                       control_message)) {
        PyObject *entry;

        /* Some platforms apparently always fill out the ancillary data
           structure with a single bogus value if none is provided; ignore it,
           if that is the case. */

        if ((!(control_message->cmsg_level)) &&
            (!(control_message->cmsg_type))) {
            continue;
        }

        /*
         * Figure out how much of the cmsg size is cmsg structure overhead - in
         * other words, how much is not part of the application data.  This lets
         * us compute the right application data size below.  There should
         * really be a CMSG_ macro for this.
         */
        cmsg_overhead = (char*)CMSG_DATA(control_message) - (char*)control_message;

        entry = Py_BuildValue(
            "(iis#)",
            control_message->cmsg_level,
            control_message->cmsg_type,
            CMSG_DATA(control_message),
            (Py_ssize_t) (control_message->cmsg_len - cmsg_overhead));
//.........这里部分代码省略.........

struct command *read_command( int cmd_fd ) {
	struct command *cmd = malloc(sizeof(struct command));
	if (! cmd) {
		fprintf(stderr, "Unable to allocate memory for command!\n");
		return NULL;
	}

	int fd[1] = {-1};
	char buffer[CMSG_SPACE(sizeof fd)];

	struct iovec v = {
		.iov_base = cmd,
		.iov_len  = sizeof(*cmd)
	};
	struct msghdr msg = {
		.msg_iov        = &v,
		.msg_iovlen     = 1,
		.msg_control    = buffer,
		.msg_controllen = sizeof(buffer)
	};

	int done = recvmsg( cmd_fd, &msg, 0 );

	if (done == -1) {
		fprintf(stderr, "Error reading command.");
		free(cmd);
		return NULL;
	}
	struct cmsghdr *cmessage = CMSG_FIRSTHDR(&msg);
	if (cmessage) {
		memcpy(fd, CMSG_DATA(cmessage), sizeof fd);
		/* place FD back in the command message */
		cmd->fd = (int) fd[0];
	}
	return cmd;
}

int send_command( int cmd_fd, enum command_type type, char *param, int passfd, int exclusive, char *tag ) {
	if (type == CMD_ADD && passfd == 1) {
		type = CMD_PASSFD;
	}
	struct command cmd = {
		.fd    = -1,
		.exclusive = exclusive,
		.type  = type,
		.param = {0},
		.tag = {0}
	};
	if (param != NULL) {
		strncpy(cmd.param, param, TH_COMMAND_PARAM_LENGTH);
		cmd.param[TH_COMMAND_PARAM_LENGTH-1] = '\0';
	}
	if (tag != NULL) {
		strncpy(cmd.tag, tag, TH_DEVICE_TAG_LENGTH);
		cmd.tag[TH_DEVICE_TAG_LENGTH-1] = '\0';
	}

	struct iovec v = {
		.iov_base = &cmd,
		.iov_len = sizeof(cmd)
	};
	struct msghdr m = {
		.msg_iov    = &v,
		.msg_iovlen = 1
	};

	/* add FD */
	int dev_fd[1] = { -1 };
	char buffer[CMSG_SPACE(sizeof(dev_fd))];
	if (passfd) {
		int fd = open( param, O_RDONLY );
		if (fd < 0) {
			perror("open");
			return 1;
		}
		dev_fd[0] = fd ;
		m.msg_control = buffer;
		m.msg_controllen = sizeof(buffer);

		struct cmsghdr *cmessage = CMSG_FIRSTHDR(&m);
		cmessage->cmsg_level = SOL_SOCKET;
		cmessage->cmsg_type = SCM_RIGHTS;
		cmessage->cmsg_len = CMSG_LEN(sizeof(dev_fd));

		m.msg_controllen = cmessage->cmsg_len;
		memcpy(CMSG_DATA(cmessage), dev_fd, sizeof dev_fd);
	}
	int done = sendmsg( cmd_fd, &m, 0 );
	return (done == -1);
}

int sendfromto(int s, void *buf, size_t len, int flags,