From patchwork Wed Nov 12 13:19:51 2008 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Patchwork-Submitter: Johann Baudy X-Patchwork-Id: 8373 X-Patchwork-Delegate: davem@davemloft.net Return-Path: X-Original-To: patchwork-incoming@ozlabs.org Delivered-To: patchwork-incoming@ozlabs.org Received: from vger.kernel.org (vger.kernel.org [209.132.176.167]) by ozlabs.org (Postfix) with ESMTP id 4E884DDDEE for ; Thu, 13 Nov 2008 00:20:31 +1100 (EST) Received: (majordomo@vger.kernel.org) by vger.kernel.org via listexpand id S1752181AbYKLNU0 (ORCPT ); Wed, 12 Nov 2008 08:20:26 -0500 Received: (majordomo@vger.kernel.org) by vger.kernel.org id S1752055AbYKLNUZ (ORCPT ); Wed, 12 Nov 2008 08:20:25 -0500 Received: from smtp3-g19.free.fr ([212.27.42.29]:56166 "EHLO smtp3-g19.free.fr" rhost-flags-OK-OK-OK-OK) by vger.kernel.org with ESMTP id S1752034AbYKLNUX (ORCPT ); Wed, 12 Nov 2008 08:20:23 -0500 Received: from smtp3-g19.free.fr (localhost.localdomain [127.0.0.1]) by smtp3-g19.free.fr (Postfix) with ESMTP id 85DCD17B56A; Wed, 12 Nov 2008 14:20:19 +0100 (CET) Received: from [192.168.16.134] (sop06-1-82-236-41-7.fbx.proxad.net [82.236.41.7]) by smtp3-g19.free.fr (Postfix) with ESMTP id 3045E17B544; Wed, 12 Nov 2008 14:20:03 +0100 (CET) Subject: Re: [PATCH] Packet socket: mmapped IO: PACKET_TX_RING From: Johann Baudy To: "Lovich, Vitali" Cc: Evgeniy Polyakov , "netdev@vger.kernel.org" Date: Wed, 12 Nov 2008 14:19:51 +0100 Message-Id: <1226495991.5444.76.camel@localhost> Mime-Version: 1.0 X-Mailer: Evolution 2.22.3.1 Sender: netdev-owner@vger.kernel.org Precedence: bulk List-ID: X-Mailing-List: netdev@vger.kernel.org Hi Vitali, Please find below my draft. According to our discussion and your proposal to replace Mutex, I've made this draft. A counter busy: That is used to prevent send from simultaneous send(). Initialized to 1, decremented at beginning of send()( with atomic_dec_and_test()), returns EBUSY if result != 0. This one is used in packet_set_ring() and tpacket_snd(). A flag shutdown: To notify send() that a shutdown is ongoing. When a release occurs, packet_release raises shutdown flag then it performs a while(atomic(busy counter != 1)); in order to wait end of send() procedure. In tpacket_send, if shutdown is seen, we exit the loop. In skb destructor, if shutdown is seen, we call sock_wfree directly(). I think such loop is better than delaying close procedure. To be sure that if close() has returned, everything is clean into the kernel. And resize is not allowed if (as we said): > - someone has mapped the socket (1.a) > - sending is ongoing - tx_pending_skb != 0 (1.b) > - a send() procedure is ongoing (1.c) I've also put pending and busy counters in packet_ring_buffer. Not useful for RX, but I think it's more clean. PS: For the moment, I'm paranoid so I've added smp_rmb/smp_wmb/and dcache flush for get/set status and also when filling skb !! (I've lost 10MBytes/s :( on my bench) - I'm using skb->mark again to forward buffer index to destructor. - I've not tested SOCK_DGRAM yet, but I've updated to_write and data contents properly (I hope) Thanks in advance, Johann Documentation/networking/packet_mmap.txt | 133 +++++++- include/linux/if_packet.h | 1 + net/packet/af_packet.c | 539 ++++++++++++++++++++++++------ 3 files changed, 557 insertions(+), 116 deletions(-) Thanks, Johann diff --git a/Documentation/networking/packet_mmap.txt b/Documentation/networking/packet_mmap.txt index 07c53d5..38e5fa3 100644 --- a/Documentation/networking/packet_mmap.txt +++ b/Documentation/networking/packet_mmap.txt @@ -4,8 +4,8 @@ This file documents the CONFIG_PACKET_MMAP option available with the PACKET socket interface on 2.4 and 2.6 kernels. This type of sockets is used for -capture network traffic with utilities like tcpdump or any other that uses -the libpcap library. +capture network traffic with utilities like tcpdump or any other that needs +raw access to network interface. You can find the latest version of this document at @@ -14,6 +14,8 @@ You can find the latest version of this document at Please send me your comments to Ulisses Alonso Camaró + Johann Baudy (TX RING) + Vitali Lovitch (TX RING) ------------------------------------------------------------------------------- + Why use PACKET_MMAP @@ -25,19 +27,24 @@ to capture each packet, it requires two if you want to get packet's timestamp (like libpcap always does). In the other hand PACKET_MMAP is very efficient. PACKET_MMAP provides a size -configurable circular buffer mapped in user space. This way reading packets just -needs to wait for them, most of the time there is no need to issue a single -system call. By using a shared buffer between the kernel and the user -also has the benefit of minimizing packet copies. - -It's fine to use PACKET_MMAP to improve the performance of the capture process, -but it isn't everything. At least, if you are capturing at high speeds (this -is relative to the cpu speed), you should check if the device driver of your -network interface card supports some sort of interrupt load mitigation or -(even better) if it supports NAPI, also make sure it is enabled. +configurable circular buffer mapped in user space that can be used to either +send or receive packets. This way reading packets just needs to wait for them, +most of the time there is no need to issue a single system call. Concerning +transmission, multiple packets can be sent through one system call to get the +highest bandwidth. +By using a shared buffer between the kernel and the user also has the benefit +of minimizing packet copies. + +It's fine to use PACKET_MMAP to improve the performance of the capture and +transmission process, but it isn't everything. At least, if you are capturing +at high speeds (this is relative to the cpu speed), you should check if the +device driver of your network interface card supports some sort of interrupt +load mitigation or (even better) if it supports NAPI, also make sure it is +enabled. For transmission, check the MTU (Maximum Transmission Unit) used and +supported by devices of your network. -------------------------------------------------------------------------------- -+ How to use CONFIG_PACKET_MMAP ++ How to use CONFIG_PACKET_MMAP to improve capture process -------------------------------------------------------------------------------- From the user standpoint, you should use the higher level libpcap library, which @@ -57,7 +64,7 @@ the low level details or want to improve libpcap by including PACKET_MMAP support. -------------------------------------------------------------------------------- -+ How to use CONFIG_PACKET_MMAP directly ++ How to use CONFIG_PACKET_MMAP directly to improve capture process -------------------------------------------------------------------------------- From the system calls stand point, the use of PACKET_MMAP involves @@ -66,6 +73,7 @@ the following process: [setup] socket() -------> creation of the capture socket setsockopt() ---> allocation of the circular buffer (ring) + option: PACKET_RX_RING mmap() ---------> mapping of the allocated buffer to the user process @@ -97,13 +105,75 @@ also the mapping of the circular buffer in the user process and the use of this buffer. -------------------------------------------------------------------------------- ++ How to use CONFIG_PACKET_MMAP directly to improve transmission process +-------------------------------------------------------------------------------- +Transmission process is similar to capture as shown below. + +[setup] socket() -------> creation of the transmission socket + setsockopt() ---> allocation of the circular buffer (ring) + option: PACKET_TX_RING + bind() ---------> bind transmission socket with a network interface + mmap() ---------> mapping of the allocated buffer to the + user process + +[transmission] poll() ---------> wait for free packets (optional) + send() ---------> send all packets that are set as ready in + the ring + The flag MSG_DONTWAIT can be used to return + before end of transfer. + +[shutdown] close() --------> destruction of the transmission socket and + deallocation of all associated resources. + +Binding the socket to your network interface is mandatory (with zero copy) to +know the header size of frames used in the circular buffer. + +As capture, each frame contains two parts: + + -------------------- +| struct tpacket_hdr | Header. It contains the status of +| | of this frame +|--------------------| +| data buffer | +. . Data that will be sent over the network interface. +. . + -------------------- + + bind() associates the socket to your network interface thanks to + sll_ifindex parameter of struct sockaddr_ll. + + Initialization example: + + struct sockaddr_ll my_addr; + struct ifreq s_ifr; + ... + + strncpy (s_ifr.ifr_name, "eth0", sizeof(s_ifr.ifr_name)); + + /* get interface index of eth0 */ + ioctl(this->socket, SIOCGIFINDEX, &s_ifr); + + /* fill sockaddr_ll struct to prepare binding */ + my_addr.sll_family = AF_PACKET; + my_addr.sll_protocol = ETH_P_ALL; + my_addr.sll_ifindex = s_ifr.ifr_ifindex; + + /* bind socket to eth0 */ + bind(this->socket, (struct sockaddr *)&my_addr, sizeof(struct sockaddr_ll)); + + A complete tutorial is available at: http://wiki.gnu-log.net/ + +-------------------------------------------------------------------------------- + PACKET_MMAP settings -------------------------------------------------------------------------------- To setup PACKET_MMAP from user level code is done with a call like + - Capture process setsockopt(fd, SOL_PACKET, PACKET_RX_RING, (void *) &req, sizeof(req)) + - Transmission process + setsockopt(fd, SOL_PACKET, PACKET_TX_RING, (void *) &req, sizeof(req)) The most significant argument in the previous call is the req parameter, this parameter must to have the following structure: @@ -117,11 +187,11 @@ this parameter must to have the following structure: }; This structure is defined in /usr/include/linux/if_packet.h and establishes a -circular buffer (ring) of unswappable memory mapped in the capture process. +circular buffer (ring) of unswappable memory. Being mapped in the capture process allows reading the captured frames and related meta-information like timestamps without requiring a system call. -Captured frames are grouped in blocks. Each block is a physically contiguous +Frames are grouped in blocks. Each block is a physically contiguous region of memory and holds tp_block_size/tp_frame_size frames. The total number of blocks is tp_block_nr. Note that tp_frame_nr is a redundant parameter because @@ -336,6 +406,7 @@ struct tpacket_hdr). If this field is 0 means that the frame is ready to be used for the kernel, If not, there is a frame the user can read and the following flags apply: ++++ Capture process: from include/linux/if_packet.h #define TP_STATUS_COPY 2 @@ -391,6 +462,36 @@ packets are in the ring: It doesn't incur in a race condition to first check the status value and then poll for frames. + +++ Transmission process +Those defines are also used for transmission: + + #define TP_STATUS_KERNEL 0 // Frame is available + #define TP_STATUS_USER 1 // Frame will be sent on next send() + #define TP_STATUS_COPY 2 // Frame is currently in transmission + +First, the kernel initializes all frames to TP_STATUS_KERNEL. To send a packet, +the user fills a data buffer of an available frame, sets tp_len to current +data buffer size and sets its status field to TP_STATUS_USER. This can be done +on multiple frames. Once the user is ready to transmit, it calls send(). +Then all buffers with status equal to TP_STATUS_USER are forwarded to the +network device. The kernel updates each status of sent frames with +TP_STATUS_COPY until the end of transfer. +At the end of each transfer, buffer status returns to TP_STATUS_KERNEL. + + header->tp_len = in_i_size; + header->tp_status = TP_STATUS_USER; + retval = send(this->socket, NULL, 0, 0); + +The user can also use poll() to check if a buffer is available: +(status == TP_STATUS_KERNEL) + + struct pollfd pfd; + pfd.fd = fd; + pfd.revents = 0; + pfd.events = POLLOUT; + retval = poll(&pfd, 1, timeout); + -------------------------------------------------------------------------------- + THANKS -------------------------------------------------------------------------------- diff --git a/include/linux/if_packet.h b/include/linux/if_packet.h index 18db066..6682963 100644 --- a/include/linux/if_packet.h +++ b/include/linux/if_packet.h @@ -46,6 +46,7 @@ struct sockaddr_ll #define PACKET_VERSION 10 #define PACKET_HDRLEN 11 #define PACKET_RESERVE 12 +#define PACKET_TX_RING 13 struct tpacket_stats { diff --git a/include/linux/skbuff.h b/include/linux/skbuff.h index 9099237..2d67a39 100644 diff --git a/net/packet/af_packet.c b/net/packet/af_packet.c old mode 100644 new mode 100755 index c718e7e..fb776a7 --- a/net/packet/af_packet.c +++ b/net/packet/af_packet.c @@ -156,7 +156,26 @@ struct packet_mreq_max }; #ifdef CONFIG_PACKET_MMAP -static int packet_set_ring(struct sock *sk, struct tpacket_req *req, int closing); +static int packet_set_ring(struct sock *sk, struct tpacket_req *req, + int closing, int tx_ring); + +struct packet_ring_buffer { + char * *pg_vec; + unsigned int head; + unsigned int frames_per_block; + unsigned int frame_size; + unsigned int frame_max; + + unsigned int pg_vec_order; + unsigned int pg_vec_pages; + unsigned int pg_vec_len; + + atomic_t busy; + atomic_t pending; +}; + +struct packet_sock; +static int tpacket_snd(struct packet_sock *po, struct msghdr *msg); #endif static void packet_flush_mclist(struct sock *sk); @@ -166,11 +185,9 @@ struct packet_sock { struct sock sk; struct tpacket_stats stats; #ifdef CONFIG_PACKET_MMAP - char * *pg_vec; - unsigned int head; - unsigned int frames_per_block; - unsigned int frame_size; - unsigned int frame_max; + struct packet_ring_buffer rx_ring; + struct packet_ring_buffer tx_ring; + atomic_t shutdown; int copy_thresh; #endif struct packet_type prot_hook; @@ -183,9 +200,6 @@ struct packet_sock { struct packet_mclist *mclist; #ifdef CONFIG_PACKET_MMAP atomic_t mapped; - unsigned int pg_vec_order; - unsigned int pg_vec_pages; - unsigned int pg_vec_len; enum tpacket_versions tp_version; unsigned int tp_hdrlen; unsigned int tp_reserve; @@ -204,36 +218,29 @@ struct packet_skb_cb { #ifdef CONFIG_PACKET_MMAP -static void *packet_lookup_frame(struct packet_sock *po, unsigned int position, - int status) +static void __packet_set_status(struct packet_sock *po, void *frame, int status) { - unsigned int pg_vec_pos, frame_offset; union { struct tpacket_hdr *h1; struct tpacket2_hdr *h2; void *raw; } h; - pg_vec_pos = position / po->frames_per_block; - frame_offset = position % po->frames_per_block; - - h.raw = po->pg_vec[pg_vec_pos] + (frame_offset * po->frame_size); + h.raw = frame; switch (po->tp_version) { case TPACKET_V1: - if (status != h.h1->tp_status ? TP_STATUS_USER : - TP_STATUS_KERNEL) - return NULL; + h.h1->tp_status = status; break; case TPACKET_V2: - if (status != h.h2->tp_status ? TP_STATUS_USER : - TP_STATUS_KERNEL) - return NULL; + h.h2->tp_status = status; break; } - return h.raw; + + smp_wmb(); + flush_dcache_page(virt_to_page(h.raw)); } -static void __packet_set_status(struct packet_sock *po, void *frame, int status) +static int __packet_get_status(struct packet_sock *po, void *frame) { union { struct tpacket_hdr *h1; @@ -241,16 +248,71 @@ static void __packet_set_status(struct packet_sock *po, void *frame, int status) void *raw; } h; + smp_rmb(); + flush_dcache_page(virt_to_page(h.raw)); + h.raw = frame; switch (po->tp_version) { case TPACKET_V1: - h.h1->tp_status = status; + return h.h1->tp_status; break; case TPACKET_V2: - h.h2->tp_status = status; + return h.h2->tp_status; break; } + return 0; +} + +static void *packet_lookup_frame(struct packet_sock *po, + struct packet_ring_buffer *buff, + unsigned int position, + int status) +{ + unsigned int pg_vec_pos, frame_offset; + union { + struct tpacket_hdr *h1; + struct tpacket2_hdr *h2; + void *raw; + } h; + + pg_vec_pos = position / buff->frames_per_block; + frame_offset = position % buff->frames_per_block; + + h.raw = buff->pg_vec[pg_vec_pos] + (frame_offset * buff->frame_size); + + if( status != __packet_get_status(po, h.raw) ) + return NULL; + + return h.raw; +} + +static inline void *packet_current_rx_frame(struct packet_sock *po, int status) +{ + return packet_lookup_frame(po, &po->rx_ring, po->rx_ring.head, status); +} + +static inline void *packet_current_tx_frame(struct packet_sock *po, int status) +{ + return packet_lookup_frame(po, &po->tx_ring, po->tx_ring.head, status); +} + +static inline void *packet_previous_rx_frame(struct packet_sock *po, int status) +{ + unsigned int previous = po->rx_ring.head ? po->rx_ring.head - 1 : po->rx_ring.frame_max; + return packet_lookup_frame(po, &po->rx_ring, previous, status); +} + +static inline void *packet_previous_tx_frame(struct packet_sock *po, int status) +{ + unsigned int previous = po->tx_ring.head ? po->tx_ring.head - 1 : po->tx_ring.frame_max; + return packet_lookup_frame(po, &po->tx_ring, previous, status); +} + +static inline void packet_increment_head(struct packet_ring_buffer *buff) +{ + buff->head = buff->head != buff->frame_max ? buff->head+1 : 0; } + #endif static inline struct packet_sock *pkt_sk(struct sock *sk) @@ -646,7 +708,7 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev, struct packe macoff = netoff - maclen; } - if (macoff + snaplen > po->frame_size) { + if (macoff + snaplen > po->rx_ring.frame_size) { if (po->copy_thresh && atomic_read(&sk->sk_rmem_alloc) + skb->truesize < (unsigned)sk->sk_rcvbuf) { @@ -659,16 +721,16 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev, struct packe if (copy_skb) skb_set_owner_r(copy_skb, sk); } - snaplen = po->frame_size - macoff; + snaplen = po->rx_ring.frame_size - macoff; if ((int)snaplen < 0) snaplen = 0; } spin_lock(&sk->sk_receive_queue.lock); - h.raw = packet_lookup_frame(po, po->head, TP_STATUS_KERNEL); + h.raw = packet_current_rx_frame(po, TP_STATUS_KERNEL); if (!h.raw) goto ring_is_full; - po->head = po->head != po->frame_max ? po->head+1 : 0; + packet_increment_head(&po->rx_ring); po->stats.tp_packets++; if (copy_skb) { status |= TP_STATUS_COPY; @@ -725,7 +787,6 @@ static int tpacket_rcv(struct sk_buff *skb, struct net_device *dev, struct packe __packet_set_status(po, h.raw, status); smp_mb(); - { struct page *p_start, *p_end; u8 *h_end = h.raw + macoff + snaplen - 1; @@ -759,10 +820,237 @@ ring_is_full: goto drop_n_restore; } -#endif +static void tpacket_destruct_skb(struct sk_buff *skb) +{ + struct packet_sock *po = pkt_sk(skb->sk); + void * ph; + BUG_ON(skb == NULL); -static int packet_sendmsg(struct kiocb *iocb, struct socket *sock, + if (unlikely(atomic_read(&po->shutdown))) + goto out_wfree; + + ph = packet_lookup_frame(po, &po->tx_ring, skb->mark, TP_STATUS_COPY); + BUG_ON(atomic_read(&po->tx_ring.pending) == 0); + atomic_dec(&po->tx_ring.pending); + __packet_set_status(po, ph, TP_STATUS_KERNEL); + +out_wfree: + sock_wfree(skb); +} + +static int tpacket_fill_skb(struct packet_sock *po, struct sk_buff * skb, void * frame, + struct net_device *dev, int size_max, __be16 proto, + unsigned char * addr) +{ + union { + struct tpacket_hdr *h1; + struct tpacket2_hdr *h2; + void *raw; + } ph; + int to_write, offset, len, tp_len, nr_frags; + struct socket *sock = po->sk.sk_socket; + struct page *page; + void *data; + int err; + + ph.raw = frame; + + skb->protocol = proto; + skb->dev = dev; + skb->priority = po->sk.sk_priority; + skb->mark = po->tx_ring.head; + + switch(po->tp_version) { + case TPACKET_V2: + tp_len = ph.h1->tp_len; + break; + default: + tp_len = ph.h1->tp_len; + break; + } + if (unlikely(tp_len > size_max)) { + printk(KERN_ERR "packet size is too long (%d > %d)\n", + tp_len, size_max); + return -EMSGSIZE; + } + + skb_reserve(skb, LL_RESERVED_SPACE(dev)); + skb_reset_network_header(skb); + + data = ph.raw + po->tp_hdrlen; + to_write = tp_len; + + if (sock->type == SOCK_DGRAM) { + err = dev_hard_header(skb, dev, ntohs(proto), addr, + NULL, tp_len); + if (unlikely(err < 0)) + return -EINVAL; + } else if (dev->hard_header_len ) { + /* net device doesn't like empty head */ + if(unlikely(tp_len <= dev->hard_header_len)) { + printk(KERN_ERR "packet size is too short " + "(%d < %d)\n", tp_len, + dev->hard_header_len); + return -EINVAL; + } + + skb_push(skb, dev->hard_header_len); + err = skb_store_bits(skb, 0, data, + dev->hard_header_len); + if (unlikely(err)) + return err; + + data += dev->hard_header_len; + to_write -= dev->hard_header_len; + } + + err = -EFAULT; + page = virt_to_page(data); + len = ((to_write > PAGE_SIZE) ? PAGE_SIZE : to_write); + + offset = (int)((long)data & (~PAGE_MASK)); + len -= offset; + + skb->data_len = to_write; + skb->len += to_write; + + while ( likely(to_write) ) { + nr_frags = skb_shinfo(skb)->nr_frags; + + if(unlikely(nr_frags >= MAX_SKB_FRAGS)) { + printk(KERN_ERR "Packet exceed the number " + "of skb frags(%lu)\n", + MAX_SKB_FRAGS); + return -EFAULT; + } + + flush_dcache_page(page); + get_page(page); + skb_fill_page_desc(skb, + nr_frags, + page++, offset, len); + to_write -= len; + len = (to_write > PAGE_SIZE) ? PAGE_SIZE : to_write; + offset = 0; + } + + return tp_len; +} + +static int tpacket_snd(struct packet_sock *po, struct msghdr *msg) +{ + struct socket *sock; + struct sk_buff *skb; + struct net_device *dev; + __be16 proto; + int ifindex, err, reserve = 0; + void * ph; + struct sockaddr_ll *saddr=(struct sockaddr_ll *)msg->msg_name; + int tp_len, size_max; + unsigned char *addr; + int len_sum = 0; + int status = 0; + + BUG_ON(po == NULL); + sock = po->sk.sk_socket; + + err = -EBUSY; + if(!atomic_dec_and_test(&po->tx_ring.busy)) + goto out; + + if (saddr == NULL) { + ifindex = po->ifindex; + proto = po->num; + addr = NULL; + } else { + err = -EINVAL; + if (msg->msg_namelen < sizeof(struct sockaddr_ll)) + goto out; + if (msg->msg_namelen < (saddr->sll_halen + offsetof(struct sockaddr_ll, sll_addr))) + goto out; + ifindex = saddr->sll_ifindex; + proto = saddr->sll_protocol; + addr = saddr->sll_addr; + } + + dev = dev_get_by_index(sock_net(&po->sk), ifindex); + err = -ENXIO; + if (unlikely(dev == NULL)) + goto out; + + err = -EINVAL; + if (unlikely(sock->type != SOCK_RAW)) + goto out_put; + + reserve = dev->hard_header_len; + + err = -ENETDOWN; + if (unlikely(!(dev->flags & IFF_UP))) + goto out_put; + + size_max = po->tx_ring.frame_size - sizeof(struct skb_shared_info) + - po->tp_hdrlen - LL_ALLOCATED_SPACE(dev); + + if (size_max > dev->mtu + reserve) + size_max = dev->mtu + reserve; + + do + { + ph = packet_current_tx_frame(po, TP_STATUS_USER); + if(unlikely(ph == NULL)) { + continue; + } + + status = TP_STATUS_USER; + skb = sock_alloc_send_skb(&po->sk, LL_ALLOCATED_SPACE(dev), + msg->msg_flags & MSG_DONTWAIT, &err); + if (unlikely(skb == NULL)) + goto out_status; + + status = TP_STATUS_LOSING; + tp_len = tpacket_fill_skb(po, skb, ph, dev, size_max, proto, + addr); + if(unlikely(tp_len < 0)) { + err = tp_len; + goto out_status; + } + + skb->destructor = tpacket_destruct_skb; + __packet_set_status(po, ph, TP_STATUS_COPY); + atomic_inc(&po->tx_ring.pending); + + status = TP_STATUS_USER; + err = dev_queue_xmit(skb); + if (unlikely(err > 0 && (err = net_xmit_errno(err)) != 0)) + goto out_xmit; + packet_increment_head(&po->tx_ring); + len_sum += tp_len; + } + while(likely((!atomic_read(&po->shutdown)) && + ((ph != NULL) + || ((!(msg->msg_flags & MSG_DONTWAIT)) + && (atomic_read(&po->tx_ring.pending)))) + )); + + err = len_sum; + goto out_put; + +out_xmit: + skb->destructor = sock_wfree; + atomic_dec(&po->tx_ring.pending); +out_status: + __packet_set_status(po, ph, status); + kfree_skb(skb); +out_put: + dev_put(dev); +out: + atomic_inc(&po->tx_ring.busy); + return err; +} +#endif + +static int packet_snd(struct socket *sock, struct msghdr *msg, size_t len) { struct sock *sk = sock->sk; @@ -853,6 +1141,19 @@ out: return err; } +static int packet_sendmsg(struct kiocb *iocb, struct socket *sock, + struct msghdr *msg, size_t len) +{ +#ifdef CONFIG_PACKET_MMAP + struct sock *sk = sock->sk; + struct packet_sock *po = pkt_sk(sk); + if (po->tx_ring.pg_vec) + return tpacket_snd(po, msg); + else +#endif + return packet_snd(sock, msg, len); +} + /* * Close a PACKET socket. This is fairly simple. We immediately go * to 'closed' state and remove our protocol entry in the device list. @@ -891,10 +1192,17 @@ static int packet_release(struct socket *sock) packet_flush_mclist(sk); #ifdef CONFIG_PACKET_MMAP - if (po->pg_vec) { + { struct tpacket_req req; memset(&req, 0, sizeof(req)); - packet_set_ring(sk, &req, 1); + atomic_set(&po->shutdown, 1); + + if (po->rx_ring.pg_vec) + packet_set_ring(sk, &req, 1, 0); + + while(unlikely(atomic_read(&po->tx_ring.busy) != 1)); + if (po->tx_ring.pg_vec) + packet_set_ring(sk, &req, 1, 1); } #endif @@ -1082,6 +1390,8 @@ static int packet_create(struct net *net, struct socket *sock, int protocol) po->running = 1; } + atomic_set(&po->tx_ring.busy, 1); + write_lock_bh(&net->packet.sklist_lock); sk_add_node(sk, &net->packet.sklist); write_unlock_bh(&net->packet.sklist_lock); @@ -1411,6 +1721,7 @@ packet_setsockopt(struct socket *sock, int level, int optname, char __user *optv #ifdef CONFIG_PACKET_MMAP case PACKET_RX_RING: + case PACKET_TX_RING: { struct tpacket_req req; @@ -1418,7 +1729,7 @@ packet_setsockopt(struct socket *sock, int level, int optname, char __user *optv return -EINVAL; if (copy_from_user(&req,optval,sizeof(req))) return -EFAULT; - return packet_set_ring(sk, &req, 0); + return packet_set_ring(sk, &req, 0, optname == PACKET_TX_RING); } case PACKET_COPY_THRESH: { @@ -1438,7 +1749,7 @@ packet_setsockopt(struct socket *sock, int level, int optname, char __user *optv if (optlen != sizeof(val)) return -EINVAL; - if (po->pg_vec) + if (po->rx_ring.pg_vec || po->tx_ring.pg_vec) return -EBUSY; if (copy_from_user(&val, optval, sizeof(val))) return -EFAULT; @@ -1457,7 +1768,7 @@ packet_setsockopt(struct socket *sock, int level, int optname, char __user *optv if (optlen != sizeof(val)) return -EINVAL; - if (po->pg_vec) + if (po->rx_ring.pg_vec || po->tx_ring.pg_vec) return -EBUSY; if (copy_from_user(&val, optval, sizeof(val))) return -EFAULT; @@ -1701,13 +2012,17 @@ static unsigned int packet_poll(struct file * file, struct socket *sock, unsigned int mask = datagram_poll(file, sock, wait); spin_lock_bh(&sk->sk_receive_queue.lock); - if (po->pg_vec) { - unsigned last = po->head ? po->head-1 : po->frame_max; - - if (packet_lookup_frame(po, last, TP_STATUS_USER)) + if (po->rx_ring.pg_vec) { + if (packet_previous_rx_frame(po, TP_STATUS_USER)) mask |= POLLIN | POLLRDNORM; } spin_unlock_bh(&sk->sk_receive_queue.lock); + spin_lock_bh(&sk->sk_write_queue.lock); + if (po->tx_ring.pg_vec) { + if (packet_current_tx_frame(po, TP_STATUS_KERNEL)) + mask |= POLLOUT | POLLWRNORM; + } + spin_unlock_bh(&sk->sk_write_queue.lock); return mask; } @@ -1783,21 +2098,34 @@ out_free_pgvec: goto out; } -static int packet_set_ring(struct sock *sk, struct tpacket_req *req, int closing) +static int packet_set_ring(struct sock *sk, struct tpacket_req *req, int closing, int tx_ring) { char **pg_vec = NULL; struct packet_sock *po = pkt_sk(sk); - int was_running, order = 0; + int was_running, busy, order = 0; + struct packet_ring_buffer *rb; + struct sk_buff_head *rb_queue; __be16 num; - int err = 0; + int err; - if (req->tp_block_nr) { - int i; + rb = tx_ring ? &po->tx_ring : &po->rx_ring; + rb_queue = tx_ring ? &sk->sk_write_queue : &sk->sk_receive_queue; - /* Sanity tests and some calculations */ + err = -EBUSY; + busy = !atomic_dec_and_test(&rb->busy); - if (unlikely(po->pg_vec)) - return -EBUSY; + if(!closing) { + if (atomic_read(&po->mapped)) + goto out; + if (busy || atomic_read(&rb->pending)) + goto out; + } + + if (req->tp_block_nr) { + /* Sanity tests and some calculations */ + err = -EBUSY; + if (unlikely(rb->pg_vec)) + goto out; switch (po->tp_version) { case TPACKET_V1: @@ -1808,42 +2136,35 @@ static int packet_set_ring(struct sock *sk, struct tpacket_req *req, int closing break; } + err = -EINVAL; if (unlikely((int)req->tp_block_size <= 0)) - return -EINVAL; + goto out; if (unlikely(req->tp_block_size & (PAGE_SIZE - 1))) - return -EINVAL; + goto out; if (unlikely(req->tp_frame_size < po->tp_hdrlen + - po->tp_reserve)) - return -EINVAL; + po->tp_reserve)) + goto out; if (unlikely(req->tp_frame_size & (TPACKET_ALIGNMENT - 1))) - return -EINVAL; + goto out; - po->frames_per_block = req->tp_block_size/req->tp_frame_size; - if (unlikely(po->frames_per_block <= 0)) - return -EINVAL; - if (unlikely((po->frames_per_block * req->tp_block_nr) != - req->tp_frame_nr)) - return -EINVAL; + rb->frames_per_block = req->tp_block_size/req->tp_frame_size; + if (unlikely(rb->frames_per_block <= 0)) + goto out; + if (unlikely((rb->frames_per_block * req->tp_block_nr) != + req->tp_frame_nr)) + goto out; err = -ENOMEM; order = get_order(req->tp_block_size); pg_vec = alloc_pg_vec(req, order); if (unlikely(!pg_vec)) goto out; - - for (i = 0; i < req->tp_block_nr; i++) { - void *ptr = pg_vec[i]; - int k; - - for (k = 0; k < po->frames_per_block; k++) { - __packet_set_status(po, ptr, TP_STATUS_KERNEL); - ptr += req->tp_frame_size; - } - } - /* Done */ - } else { + } + /* Done */ + else { + err = -EINVAL; if (unlikely(req->tp_frame_nr)) - return -EINVAL; + goto out; } lock_sock(sk); @@ -1866,20 +2187,19 @@ static int packet_set_ring(struct sock *sk, struct tpacket_req *req, int closing if (closing || atomic_read(&po->mapped) == 0) { err = 0; #define XC(a, b) ({ __typeof__ ((a)) __t; __t = (a); (a) = (b); __t; }) - - spin_lock_bh(&sk->sk_receive_queue.lock); - pg_vec = XC(po->pg_vec, pg_vec); - po->frame_max = (req->tp_frame_nr - 1); - po->head = 0; - po->frame_size = req->tp_frame_size; - spin_unlock_bh(&sk->sk_receive_queue.lock); - - order = XC(po->pg_vec_order, order); - req->tp_block_nr = XC(po->pg_vec_len, req->tp_block_nr); - - po->pg_vec_pages = req->tp_block_size/PAGE_SIZE; - po->prot_hook.func = po->pg_vec ? tpacket_rcv : packet_rcv; - skb_queue_purge(&sk->sk_receive_queue); + spin_lock_bh(&rb_queue->lock); + pg_vec = XC(rb->pg_vec, pg_vec); + rb->frame_max = (req->tp_frame_nr - 1); + rb->head = 0; + rb->frame_size = req->tp_frame_size; + spin_unlock_bh(&rb_queue->lock); + + order = XC(rb->pg_vec_order, order); + req->tp_block_nr = XC(rb->pg_vec_len, req->tp_block_nr); + + rb->pg_vec_pages = req->tp_block_size/PAGE_SIZE; + po->prot_hook.func = (po->rx_ring.pg_vec) ? tpacket_rcv : packet_rcv; + skb_queue_purge(rb_queue); #undef XC if (atomic_read(&po->mapped)) printk(KERN_DEBUG "packet_mmap: vma is busy: %d\n", atomic_read(&po->mapped)); @@ -1898,7 +2218,10 @@ static int packet_set_ring(struct sock *sk, struct tpacket_req *req, int closing if (pg_vec) free_pg_vec(pg_vec, order, req->tp_block_nr); + + err = 0; out: + atomic_inc(&rb->busy); return err; } @@ -1906,7 +2229,8 @@ static int packet_mmap(struct file *file, struct socket *sock, struct vm_area_st { struct sock *sk = sock->sk; struct packet_sock *po = pkt_sk(sk); - unsigned long size; + unsigned long size, expected_size; + struct packet_ring_buffer *rb; unsigned long start; int err = -EINVAL; int i; @@ -1917,23 +2241,38 @@ static int packet_mmap(struct file *file, struct socket *sock, struct vm_area_st size = vma->vm_end - vma->vm_start; lock_sock(sk); - if (po->pg_vec == NULL) + + expected_size = 0; + if (po->rx_ring.pg_vec) + expected_size += po->rx_ring.pg_vec_len * po->rx_ring.pg_vec_pages * PAGE_SIZE; + if (po->tx_ring.pg_vec) + expected_size += po->tx_ring.pg_vec_len * po->tx_ring.pg_vec_pages * PAGE_SIZE; + + if (expected_size == 0) goto out; - if (size != po->pg_vec_len*po->pg_vec_pages*PAGE_SIZE) + + if (size != expected_size) goto out; start = vma->vm_start; - for (i = 0; i < po->pg_vec_len; i++) { - struct page *page = virt_to_page(po->pg_vec[i]); - int pg_num; - - for (pg_num = 0; pg_num < po->pg_vec_pages; pg_num++, page++) { - err = vm_insert_page(vma, start, page); - if (unlikely(err)) - goto out; - start += PAGE_SIZE; + + for(rb = &po->rx_ring; rb <= &po->tx_ring; rb++) { + if (rb->pg_vec == NULL) + continue; + + for (i = 0; i < rb->pg_vec_len; i++) { + struct page *page = virt_to_page(rb->pg_vec[i]); + int pg_num; + + for (pg_num = 0; pg_num < rb->pg_vec_pages; pg_num++, page++) { + err = vm_insert_page(vma, start, page); + if (unlikely(err)) + goto out; + start += PAGE_SIZE; + } } } + atomic_inc(&po->mapped); vma->vm_ops = &packet_mmap_ops; err = 0;