This patch series follows up on the RFC and subsequent review comments
at https://patchwork.ozlabs.org/cover/862248/
Review comments addressed are
- drop MSG_PEEK change for sk_error_queue
- (patch4) batch of SO_EE_ORIGIN_MAX_ZCOOKIES (#defined to 8) is sent up
as part of the data in the error notification. The ancillary data in
with this notification specifies the number of cookies in ee_data,
with the ee_origin is set to SO_EE_ORIGIN_ZCOOKIE
- (patch4, patch5) allocate the skb to be used for error notification
up-front (in rds_sendmsg()) so that we never have to fail due to skb
allocation failure in the callback routine.
- other minor review fixes around refactoring code for the setsockopt
of ZEROCOPY, use iov_iter_npages() etc.
This patch series also updates the selftests/net/msg_zerocopy.c to support
PF_RDS sockets (both with and without zerocopy)
Thanks to Willem de Bruijn and Eric Dumazet for review comments.
Sowmini Varadhan (7):
skbuff: export mm_[un]account_pinned_pages for other modules
rds: hold a sock ref from rds_message to the rds_sock
sock: permit SO_ZEROCOPY on PF_RDS socket
rds: support for zcopy completion notification
rds: zerocopy Tx support.
selftests/net: add support for PF_RDS sockets
selftests/net: add zerocopy support for PF_RDS test case
include/linux/skbuff.h | 3 +
include/uapi/linux/errqueue.h | 2 +
include/uapi/linux/rds.h | 1 +
net/core/skbuff.c | 6 +-
net/core/sock.c | 25 ++--
net/rds/af_rds.c | 7 +
net/rds/message.c | 148 ++++++++++++++++++++++-
net/rds/rds.h | 23 ++++-
net/rds/recv.c | 2 +
net/rds/send.c | 51 ++++++--
tools/testing/selftests/net/msg_zerocopy.c | 182 +++++++++++++++++++++++-----
11 files changed, 390 insertions(+), 60 deletions(-)
allow the application to set SO_ZEROCOPY on the underlying sk
of a PF_RDS socket
Signed-off-by: Sowmini Varadhan <redacted>
---
net/core/sock.c | 25 ++++++++++++++-----------
1 files changed, 14 insertions(+), 11 deletions(-)
@@ -1049,18 +1049,21 @@ int sock_setsockopt(struct socket *sock, int level, int optname,break;caseSO_ZEROCOPY:-if(sk->sk_family!=PF_INET&&sk->sk_family!=PF_INET6)+if(sk->sk_family==PF_INET||sk->sk_family==PF_INET6){+if(sk->sk_protocol!=IPPROTO_TCP)+ret=-ENOTSUPP;+elseif(sk->sk_state!=TCP_CLOSE)+ret=-EBUSY;+}elseif(sk->sk_family!=PF_RDS){ret=-ENOTSUPP;-elseif(sk->sk_protocol!=IPPROTO_TCP)-ret=-ENOTSUPP;-elseif(sk->sk_state!=TCP_CLOSE)-ret=-EBUSY;-elseif(val<0||val>1)-ret=-EINVAL;-else-sock_valbool_flag(sk,SOCK_ZEROCOPY,valbool);-break;-+}+if(!ret){+if(val<0||val>1)+ret=-EINVAL;+else+sock_valbool_flag(sk,SOCK_ZEROCOPY,valbool);+break;+}default:ret=-ENOPROTOOPT;break;
The existing model holds a reference from the rds_sock to the
rds_message, but the rds_message does not itself hold a sock_put()
on the rds_sock. Instead the m_rs field in the rds_message is
assigned when the message is queued on the sock, and nulled when
the message is dequeued from the sock.
We want to be able to notify userspace when the rds_message
is actually freed (from rds_message_purge(), after the refcounts
to the rds_message go to 0). At the time that rds_message_purge()
is called, the message is no longer on the rds_sock retransmit
queue. Thus the explicit reference for the m_rs is needed to
send a notification that will signal to userspace that
it is now safe to free/reuse any pages that may have
been pinned down for zerocopy.
This patch manages the m_rs assignment in the rds_message with
the necessary refcount book-keeping.
Signed-off-by: Sowmini Varadhan <redacted>
---
net/rds/message.c | 8 +++++++-
net/rds/send.c | 7 +------
2 files changed, 8 insertions(+), 7 deletions(-)
@@ -849,6 +843,7 @@ static int rds_send_queue_rm(struct rds_sock *rs, struct rds_connection *conn,list_add_tail(&rm->m_sock_item,&rs->rs_send_queue);set_bit(RDS_MSG_ON_SOCK,&rm->m_flags);rds_message_addref(rm);+sock_hold(rds_rs_to_sk(rs));rm->m_rs=rs;/* The code ordering is a little weird, but we're
@@ -300,10 +304,15 @@ static int do_setup_tx(int domain, int type, int protocol)if(cfg_zerocopy)do_setsockopt(fd,SOL_SOCKET,SO_ZEROCOPY,1);-if(domain!=PF_PACKET)+if(domain!=PF_PACKET&&domain!=PF_RDS)if(connect(fd,(void*)&cfg_dst_addr,cfg_alen))error(1,errno,"connect");+if(domain==PF_RDS){+if(bind(fd,(void*)&cfg_src_addr,cfg_alen))+error(1,errno,"bind");+}+returnfd;}
@@ -444,6 +453,13 @@ static void do_tx(int domain, int type, int protocol)msg.msg_iovlen++;}+if(domain==PF_RDS){+msg.msg_name=&cfg_dst_addr;+msg.msg_namelen=(cfg_dst_addr.ss_family==AF_INET?+sizeof(structsockaddr_in):+sizeof(structsockaddr_in6));+}+iov[2].iov_base=payload;iov[2].iov_len=cfg_payload_len;msg.msg_iovlen++;
@@ -555,6 +571,40 @@ static void do_flush_datagram(int fd, int type)bytes+=cfg_payload_len;}++staticvoiddo_recvmsg(intfd)+{+intret,off=0;+char*buf;+structioveciov;+structmsghdrmsg;+structsockaddr_storagedin;++buf=calloc(cfg_payload_len,sizeof(char));+iov.iov_base=buf;+iov.iov_len=cfg_payload_len;++memset(&msg,0,sizeof(msg));+msg.msg_name=&din;+msg.msg_namelen=sizeof(din);+msg.msg_iov=&iov;+msg.msg_iovlen=1;++ret=recvmsg(fd,&msg,MSG_TRUNC);++if(ret==-1)+error(1,errno,"recv");+if(ret!=cfg_payload_len)+error(1,0,"recv: ret=%u != %u",ret,cfg_payload_len);++if(memcmp(buf+off,payload,ret))+error(1,0,"recv: data mismatch");++free(buf);+packets++;+bytes+=cfg_payload_len;+}+staticvoiddo_rx(intdomain,inttype,intprotocol){uint64_ttstop;
@@ -566,6 +616,8 @@ static void do_rx(int domain, int type, int protocol)do{if(type==SOCK_STREAM)do_flush_tcp(fd);+elseif(domain==PF_RDS)+do_recvmsg(fd);elsedo_flush_datagram(fd,type);
Send a cookie with sendmsg() on PF_RDS sockets, and process the
returned batched cookies in do_recv_completion()
Signed-off-by: Sowmini Varadhan <redacted>
---
tools/testing/selftests/net/msg_zerocopy.c | 119 ++++++++++++++++++++-------
1 files changed, 88 insertions(+), 31 deletions(-)
@@ -346,36 +382,57 @@ static bool do_recv_completion(int fd)cm->cmsg_level,cm->cmsg_type);serr=(void*)CMSG_DATA(cm);-if(serr->ee_origin!=SO_EE_ORIGIN_ZEROCOPY)-error(1,0,"serr: wrong origin: %u",serr->ee_origin);-if(serr->ee_errno!=0)-error(1,0,"serr: wrong error code: %u",serr->ee_errno);-hi=serr->ee_data;-lo=serr->ee_info;-range=hi-lo+1;+switch(serr->ee_origin){+caseSO_EE_ORIGIN_ZEROCOPY:{+if(serr->ee_errno!=0)+error(1,0,"serr: wrong error code: %u",+serr->ee_errno);+hi=serr->ee_data;+lo=serr->ee_info;+range=hi-lo+1;++/* Detect notification gaps. These should not happen often,+*ifatall.Gapscanoccurduetodrops,reorderingand+*retransmissions.+*/+if(lo!=next_completion)+fprintf(stderr,"gap: %u..%u does not append to %u\n",+lo,hi,next_completion);+next_completion=hi+1;++zerocopy=!(serr->ee_code&SO_EE_CODE_ZEROCOPY_COPIED);+if(zerocopied==-1)+zerocopied=zerocopy;+elseif(zerocopied!=zerocopy){+fprintf(stderr,"serr: inconsistent\n");+zerocopied=zerocopy;+}+if(cfg_verbose>=2)+fprintf(stderr,"completed: %u (h=%u l=%u)\n",+range,hi,lo);-/* Detect notification gaps. These should not happen often, if at all.-*Gapscanoccurduetodrops,reorderingandretransmissions.-*/-if(lo!=next_completion)-fprintf(stderr,"gap: %u..%u does not append to %u\n",-lo,hi,next_completion);-next_completion=hi+1;--zerocopy=!(serr->ee_code&SO_EE_CODE_ZEROCOPY_COPIED);-if(zerocopied==-1)-zerocopied=zerocopy;-elseif(zerocopied!=zerocopy){-fprintf(stderr,"serr: inconsistent\n");-zerocopied=zerocopy;+completions+=range;+break;+}+caseSO_EE_ORIGIN_ZCOOKIE:{+intncookies,i;++if(serr->ee_errno!=0)+error(1,0,"serr: wrong error code: %u",+serr->ee_errno);+ncookies=serr->ee_data;+for(i=0;i<ncookies;i++)+if(cfg_verbose>=2)+fprintf(stderr,"%d\n",ckbuf[i]);+completions+=ncookies;+zerocopied=1;+break;+}+default:+error(1,0,"serr: wrong origin: %u",serr->ee_origin);}-if(cfg_verbose>=2)-fprintf(stderr,"completed: %u (h=%u l=%u)\n",-range,hi,lo);--completions+=range;returntrue;}
@@ -470,7 +527,7 @@ static void do_tx(int domain, int type, int protocol)if(cfg_cork)do_sendmsg_corked(fd,&msg);else-do_sendmsg(fd,&msg,cfg_zerocopy);+do_sendmsg(fd,&msg,cfg_zerocopy,domain);while(!do_poll(fd,POLLOUT)){if(cfg_zerocopy)
If the MSG_ZEROCOPY flag is specified with rds_sendmsg(), and,
if the SO_ZEROCOPY socket option has been set on the PF_RDS socket,
application pages sent down with rds_sendmsg() are pinned.
The pinning uses the accounting infrastructure added by
Commit a91dbff551a6 ("sock: ulimit on MSG_ZEROCOPY pages")
The payload bytes in the message may not be modified for the
duration that the message has been pinned. A multi-threaded
application using this infrastructure may thus need to be notified
about send-completion so that it can free/reuse the buffers
passed to rds_sendmsg(). Notification of send-completion will
identify each message-buffer by a cookie that the application
must specify as ancillary data to rds_sendmsg().
The ancillary data in this case has cmsg_level == SOL_RDS
and cmsg_type == RDS_CMSG_ZCOPY_COOKIE.
Signed-off-by: Sowmini Varadhan <redacted>
---
include/uapi/linux/rds.h | 1 +
net/rds/message.c | 48 +++++++++++++++++++++++++++++++++++++++++++++-
net/rds/rds.h | 3 +-
net/rds/send.c | 44 ++++++++++++++++++++++++++++++++++++-----
4 files changed, 88 insertions(+), 8 deletions(-)
@@ -899,6 +900,8 @@ static int rds_rm_size(struct msghdr *msg, int data_len)break;+caseRDS_CMSG_ZCOPY_COOKIE:+zcopy_cookie=true;caseRDS_CMSG_RDMA_DEST:caseRDS_CMSG_RDMA_MAP:cmsg_groups|=2;
@@ -919,7 +922,10 @@ static int rds_rm_size(struct msghdr *msg, int data_len)}-size+=ceil(data_len,PAGE_SIZE)*sizeof(structscatterlist);+if((msg->msg_flags&MSG_ZEROCOPY)&&!zcopy_cookie)+return-EINVAL;++size+=num_sgs*sizeof(structscatterlist);/* Ensure (DEST, MAP) are never used with (ARGS, ATOMIC) */if(cmsg_groups==3)
@@ -928,6 +934,18 @@ static int rds_rm_size(struct msghdr *msg, int data_len)returnsize;}+staticintrds_cmsg_zcopy(structrds_sock*rs,structrds_message*rm,+structcmsghdr*cmsg)+{+u32*cookie;++if(cmsg->cmsg_len<CMSG_LEN(sizeof(*cookie)))+return-EINVAL;+cookie=CMSG_DATA(cmsg);+rm->data.op_mmp_znotifier->z_cookie=*cookie;+return0;+}+staticintrds_cmsg_send(structrds_sock*rs,structrds_message*rm,structmsghdr*msg,int*allocated_mr){
@@ -1040,10 +1062,13 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)longtimeo=sock_sndtimeo(sk,nonblock);structrds_conn_path*cpath;size_ttotal_payload_len=payload_len,rdma_payload_len=0;+boolzcopy=((msg->msg_flags&MSG_ZEROCOPY)&&+sock_flag(rds_rs_to_sk(rs),SOCK_ZEROCOPY));+intnum_sgs=ceil(payload_len,PAGE_SIZE);/* Mirror Linux UDP mirror of BSD error message compatibility *//* XXX: Perhaps MSG_MORE someday */-if(msg->msg_flags&~(MSG_DONTWAIT|MSG_CMSG_COMPAT)){+if(msg->msg_flags&~(MSG_DONTWAIT|MSG_CMSG_COMPAT|MSG_ZEROCOPY)){ret=-EOPNOTSUPP;gotoout;}
@@ -1087,8 +1112,15 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)gotoout;}+if(zcopy){+if(rs->rs_transport->t_type!=RDS_TRANS_TCP){+ret=-EOPNOTSUPP;+gotoout;+}+num_sgs=iov_iter_npages(&msg->msg_iter,INT_MAX);+}/* size of rm including all sgs */-ret=rds_rm_size(msg,payload_len);+ret=rds_rm_size(msg,payload_len,num_sgs);if(ret<0)gotoout;
@@ -1100,12 +1132,12 @@ int rds_sendmsg(struct socket *sock, struct msghdr *msg, size_t payload_len)/* Attach data to the rm */if(payload_len){-rm->data.op_sg=rds_message_alloc_sgs(rm,ceil(payload_len,PAGE_SIZE));+rm->data.op_sg=rds_message_alloc_sgs(rm,num_sgs);if(!rm->data.op_sg){ret=-ENOMEM;gotoout;}-ret=rds_message_copy_from_user(rm,&msg->msg_iter);+ret=rds_message_copy_from_user(rm,&msg->msg_iter,zcopy);if(ret)gotoout;}
RDS removes a datagram (rds_message) from the retransmit queue when
an ACK is received. The ACK indicates that the receiver has queued
the RDS datagram, so that the sender can safely forget the datagram.
When all references to the rds_message are quiesced, rds_message_purge
is called to release resources used by the rds_message
If the datagram to be removed had pinned pages set up, add
an entry to the rs->rs_znotify_queue so that the notifcation
will be sent up via rds_rm_zerocopy_callback() when the
rds_message is eventually freed by rds_message_purge.
rds_rm_zerocopy_callback() attempts to batch the number of cookies
sent with each notification to a max of SO_EE_ORIGIN_MAX_ZCOOKIES.
Each time a cookie is released by rds_message_purge(), the
rs_znotify_queue is checked to see if the MAX_ZCOOKIES batch limit
has been exceeded (in which case we send up a notification). If the
limit has not been exceeded, the cookie is added to the rs_znotify_queue
and a timer is set up, to make sure the cookie notification will
be sent after an upper bound of RDS_REAP_TIMEOUT (should the
traffic rate slow down)
Signed-off-by: Sowmini Varadhan <redacted>
---
include/uapi/linux/errqueue.h | 2 +
net/rds/af_rds.c | 7 +++
net/rds/message.c | 104 ++++++++++++++++++++++++++++++++++++++---
net/rds/rds.h | 20 ++++++++
net/rds/recv.c | 2 +
5 files changed, 128 insertions(+), 7 deletions(-)
@@ -183,6 +185,8 @@ static unsigned int rds_poll(struct file *file, struct socket *sock,mask|=(POLLIN|POLLRDNORM);if(rs->rs_snd_bytes<rds_sk_sndbuf(rs))mask|=(POLLOUT|POLLWRNORM);+if(sk->sk_err||!skb_queue_empty(&sk->sk_error_queue))+mask|=POLLERR;read_unlock_irqrestore(&rs->rs_recv_lock,flags);/* clear state any time we wake a seen-congested socket */
@@ -511,6 +515,9 @@ static int __rds_create(struct socket *sock, struct sock *sk, int protocol)INIT_LIST_HEAD(&rs->rs_send_queue);INIT_LIST_HEAD(&rs->rs_recv_queue);INIT_LIST_HEAD(&rs->rs_notify_queue);+INIT_LIST_HEAD(&rs->rs_znotify_queue);+rs->rs_ncookies=0;+timer_setup(&rs->rs_cookie_timer,rs_zcopy_notify,0);INIT_LIST_HEAD(&rs->rs_cong_list);spin_lock_init(&rs->rs_rdma_lock);rs->rs_rdma_keys=RB_ROOT;
@@ -53,28 +56,115 @@ void rds_message_addref(struct rds_message *rm)}EXPORT_SYMBOL_GPL(rds_message_addref);+staticvoidrds_rm_zerocopy_callback(structrds_sock*rs,+structrds_znotifier*znotifier,+boolforce)+{+structsock*sk=rds_rs_to_sk(rs);+structsk_buff*skb;+structsock_exterr_skb*serr;+unsignedlongflags;+u32*ptr;+intncookies=0,i;+structrds_znotifier*znotif,*ztmp,*first;+LIST_HEAD(tmp_list);++spin_lock_irqsave(&rs->rs_lock,flags);+ncookies=rs->rs_ncookies;+if(ncookies<SO_EE_ORIGIN_MAX_ZCOOKIES&&!force){+if(znotifier){/* add this cookie to the list and return */+list_add_tail(&znotifier->z_list,+&rs->rs_znotify_queue);+rs->rs_ncookies++;+}+spin_unlock_irqrestore(&rs->rs_lock,flags);+return;+}+if(!ncookies){/* timer finds a reaped list */+spin_unlock_irqrestore(&rs->rs_lock,flags);+return;+}+/* reap existing cookie list if we have hit the max, then add+*newcookietothelistfornextroundofreaping.+*/+list_splice(&rs->rs_znotify_queue,&tmp_list);/* reap now */+INIT_LIST_HEAD(&rs->rs_znotify_queue);+rs->rs_ncookies=0;+if(znotifier){/* for next round */+list_add_tail(&znotifier->z_list,&rs->rs_znotify_queue);+rs->rs_ncookies++;+}+spin_unlock_irqrestore(&rs->rs_lock,flags);++first=list_first_entry(&tmp_list,structrds_znotifier,z_list);+znotif=list_next_entry(first,z_list);+list_del(&first->z_list);++skb=rds_skb_from_znotifier(first);+ptr=skb_put(skb,ncookies*sizeof(u32));+i=0;+ptr[i++]=first->z_cookie;++list_for_each_entry_safe(znotif,ztmp,&tmp_list,z_list){+list_del(&znotif->z_list);+ptr[i++]=znotif->z_cookie;+mm_unaccount_pinned_pages(&znotif->z_mmp);+consume_skb(rds_skb_from_znotifier(znotif));+}+WARN_ON(!list_empty(&tmp_list));++serr=SKB_EXT_ERR(skb);+serr->ee.ee_errno=0;+serr->ee.ee_origin=SO_EE_ORIGIN_ZCOOKIE;+serr->ee.ee_data=ncookies;+serr->ee.ee_info=0;+serr->ee.ee_code|=SO_EE_CODE_ZEROCOPY_COPIED;++if(sock_queue_err_skb(sk,skb))+consume_skb(skb);+}++voidrs_zcopy_notify(structtimer_list*t)+{+structrds_sock*rs=from_timer(rs,t,rs_cookie_timer);++rds_rm_zerocopy_callback(rs,NULL,true);+}+/**Thisreliesondma_map_sg()nottouchingsg[].pageduringmerging.*/staticvoidrds_message_purge(structrds_message*rm){unsignedlongi,flags;+boolzcopy=false;if(unlikely(test_bit(RDS_MSG_PAGEVEC,&rm->m_flags)))return;+spin_lock_irqsave(&rm->m_rs_lock,flags);+if(rm->data.op_mmp_znotifier&&rm->m_rs){+structrds_sock*rs=rm->m_rs;++zcopy=true;+rds_rm_zerocopy_callback(rs,rm->data.op_mmp_znotifier,false);+rm->data.op_mmp_znotifier=NULL;+(void)mod_timer(&rs->rs_cookie_timer,RDS_REAP_TIMEOUT);++sock_put(rds_rs_to_sk(rs));+rm->m_rs=NULL;+}+spin_unlock_irqrestore(&rm->m_rs_lock,flags);+for(i=0;i<rm->data.op_nents;i++){rdsdebug("putting data page %p\n",(void*)sg_page(&rm->data.op_sg[i]));/* XXX will have to put_page for page refs */-__free_page(sg_page(&rm->data.op_sg[i]));+if(!zcopy)+__free_page(sg_page(&rm->data.op_sg[i]));+else+put_page(sg_page(&rm->data.op_sg[i]));}rm->data.op_nents=0;-spin_lock_irqsave(&rm->m_rs_lock,flags);-if(rm->m_rs){-sock_put(rds_rs_to_sk(rm->m_rs));-rm->m_rs=NULL;-}-spin_unlock_irqrestore(&rm->m_rs_lock,flags);if(rm->rdma.op_active)rds_rdma_free_op(&rm->rdma);
@@ -594,6 +594,8 @@ int rds_recvmsg(struct socket *sock, struct msghdr *msg, size_t size,if(msg_flags&MSG_OOB)gotoout;+if(msg_flags&MSG_ERRQUEUE)+returnsock_recv_errqueue(sk,msg,size,SOL_IP,IP_RECVERR);while(1){/* If there are pending notifications, do those - and nothing else */
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-25 14:45:34
On Wed, Jan 24, 2018 at 12:45 PM, Sowmini Varadhan
[off-list ref] wrote:
The existing model holds a reference from the rds_sock to the
rds_message, but the rds_message does not itself hold a sock_put()
on the rds_sock. Instead the m_rs field in the rds_message is
assigned when the message is queued on the sock, and nulled when
the message is dequeued from the sock.
We want to be able to notify userspace when the rds_message
is actually freed (from rds_message_purge(), after the refcounts
to the rds_message go to 0). At the time that rds_message_purge()
is called, the message is no longer on the rds_sock retransmit
queue. Thus the explicit reference for the m_rs is needed to
send a notification that will signal to userspace that
it is now safe to free/reuse any pages that may have
been pinned down for zerocopy.
This patch manages the m_rs assignment in the rds_message with
the necessary refcount book-keeping.
You may alos be able to do the same as tcp zerocopy and
hold an sk reference on the notification skb.
This patch manages the m_rs assignment in the rds_message with
the necessary refcount book-keeping.
You may alos be able to do the same as tcp zerocopy and
hold an sk reference on the notification skb.
We tether the notification skb to the rds socket after the
refcount on the rds_message goes to zero, so we already have a
ref from the sk to the notification skb,
If we kept a refcount of all notification skb's (even the ones that
are not ready to be unpinned yet) on the sk, then we have additional
complexity trying to figure out which skb's are ready for notification
at any point, so not sure it would make things simpler..
--Sowmini
Hi Sowmini,
On 1/24/2018 3:45 AM, Sowmini Varadhan wrote:
This patch series follows up on the RFC and subsequent review comments
at https://patchwork.ozlabs.org/cover/862248/
Review comments addressed are
- drop MSG_PEEK change for sk_error_queue
- (patch4) batch of SO_EE_ORIGIN_MAX_ZCOOKIES (#defined to 8) is sent up
as part of the data in the error notification. The ancillary data in
with this notification specifies the number of cookies in ee_data,
with the ee_origin is set to SO_EE_ORIGIN_ZCOOKIE
- (patch4, patch5) allocate the skb to be used for error notification
up-front (in rds_sendmsg()) so that we never have to fail due to skb
allocation failure in the callback routine.
- other minor review fixes around refactoring code for the setsockopt
of ZEROCOPY, use iov_iter_npages() etc.
This patch series also updates the selftests/net/msg_zerocopy.c to support
PF_RDS sockets (both with and without zerocopy)
RDS changes looks like largely good but I need some time to look at the
completion notification and send side changes. Will try to provide
feedback in next few days.
regards,
Santosh
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-28 13:52:30
On Thu, Jan 25, 2018 at 4:35 PM, Sowmini Varadhan
[off-list ref] wrote:
On (01/25/18 15:44), Willem de Bruijn wrote:
quoted
quoted
This patch manages the m_rs assignment in the rds_message with
the necessary refcount book-keeping.
You may alos be able to do the same as tcp zerocopy and
hold an sk reference on the notification skb.
We tether the notification skb to the rds socket after the
refcount on the rds_message goes to zero, so we already have a
ref from the sk to the notification skb,
If we kept a refcount of all notification skb's (even the ones that
are not ready to be unpinned yet) on the sk, then we have additional
complexity trying to figure out which skb's are ready for notification
at any point, so not sure it would make things simpler..
I don't quite follow. Every notification skb is created when pages refcount
is increased. It persists until at least rds_rm_zerocopy_callback, after data
skb has been freed and pages refcount has been decreased.
In this callback, skb is consumed if another skb is already queued on
the error queue, otherwise it is queued itself. It needs to hold a sock ref
until it can be queued.
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-28 13:57:35
On Wed, Jan 24, 2018 at 12:45 PM, Sowmini Varadhan
[off-list ref] wrote:
RDS removes a datagram (rds_message) from the retransmit queue when
an ACK is received. The ACK indicates that the receiver has queued
the RDS datagram, so that the sender can safely forget the datagram.
When all references to the rds_message are quiesced, rds_message_purge
is called to release resources used by the rds_message
If the datagram to be removed had pinned pages set up, add
an entry to the rs->rs_znotify_queue so that the notifcation
will be sent up via rds_rm_zerocopy_callback() when the
rds_message is eventually freed by rds_message_purge.
rds_rm_zerocopy_callback() attempts to batch the number of cookies
sent with each notification to a max of SO_EE_ORIGIN_MAX_ZCOOKIES.
Each time a cookie is released by rds_message_purge(), the
rs_znotify_queue is checked to see if the MAX_ZCOOKIES batch limit
has been exceeded (in which case we send up a notification). If the
limit has not been exceeded, the cookie is added to the rs_znotify_queue
and a timer is set up
An alternative that does not require a timer is to batch on the sk
error queue itself, like tcp zerocopy. That queues the first notification
skb on the error queue without any notification latency.
Then, if a subsequent notification comes in while another is pending
with < MAX zcookies, it coalesces the new notification onto the pending
skb and consumes the other. For RDS notifications, the implementation
is an extra skb_put + uint32_t assignment.
Optionally, the socket can trigger another sk_error_report on each
new notification.
+static void rds_rm_zerocopy_callback(struct rds_sock *rs,
+ struct rds_znotifier *znotifier,
+ bool force)
+{
+ struct sock *sk = rds_rs_to_sk(rs);
+ struct sk_buff *skb;
+ struct sock_exterr_skb *serr;
+ unsigned long flags;
+ u32 *ptr;
+ int ncookies = 0, i;
+ struct rds_znotifier *znotif, *ztmp, *first;
+ LIST_HEAD(tmp_list);
+
+ spin_lock_irqsave(&rs->rs_lock, flags);
+ ncookies = rs->rs_ncookies;
+ if (ncookies < SO_EE_ORIGIN_MAX_ZCOOKIES && !force) {
+ if (znotifier) { /* add this cookie to the list and return */
can be checked before taking lock.
More importantly, when is this ever NULL? This function is a callback
for a zerocopy struct of type znotifier. Is it doing double duty to flush
any outstanding if znotifier == NULL && force == true? If so, the first
condition probably never occurs unless force == true and thus the
second is redundant.
+ list_add_tail(&znotifier->z_list,
+ &rs->rs_znotify_queue);
+ rs->rs_ncookies++;
+ }
+ spin_unlock_irqrestore(&rs->rs_lock, flags);
+ return;
+ }
+ if (!ncookies) { /* timer finds a reaped list */
+ spin_unlock_irqrestore(&rs->rs_lock, flags);
+ return;
+ }
+ /* reap existing cookie list if we have hit the max, then add
+ * new cookie to the list for next round of reaping.
+ */
+ list_splice(&rs->rs_znotify_queue, &tmp_list); /* reap now */
+ INIT_LIST_HEAD(&rs->rs_znotify_queue);
+ rs->rs_ncookies = 0;
+ if (znotifier) { /* for next round */
This adds unnecessary notification latency to delivery of current
notification. The latest notification can be appended to tmp_list and
sent up immediately.
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-28 13:58:17
On Wed, Jan 24, 2018 at 12:46 PM, Sowmini Varadhan
[off-list ref] wrote:
quoted hunk
If the MSG_ZEROCOPY flag is specified with rds_sendmsg(), and,
if the SO_ZEROCOPY socket option has been set on the PF_RDS socket,
application pages sent down with rds_sendmsg() are pinned.
The pinning uses the accounting infrastructure added by
Commit a91dbff551a6 ("sock: ulimit on MSG_ZEROCOPY pages")
The payload bytes in the message may not be modified for the
duration that the message has been pinned. A multi-threaded
application using this infrastructure may thus need to be notified
about send-completion so that it can free/reuse the buffers
passed to rds_sendmsg(). Notification of send-completion will
identify each message-buffer by a cookie that the application
must specify as ancillary data to rds_sendmsg().
The ancillary data in this case has cmsg_level == SOL_RDS
and cmsg_type == RDS_CMSG_ZCOPY_COOKIE.
Signed-off-by: Sowmini Varadhan <redacted>
---
include/uapi/linux/rds.h | 1 +
net/rds/message.c | 48 +++++++++++++++++++++++++++++++++++++++++++++-
net/rds/rds.h | 3 +-
net/rds/send.c | 44 ++++++++++++++++++++++++++++++++++++-----
4 files changed, 88 insertions(+), 8 deletions(-)
@@ -377,6 +379,50 @@ int rds_message_copy_from_user(struct rds_message *rm, struct iov_iter *from)sg=rm->data.op_sg;sg_off=0;/* Dear gcc, sg->page will be null from kzalloc. */+if(zcopy){+inttotal_copied=0;+structsk_buff*skb;++skb=alloc_skb(SO_EE_ORIGIN_MAX_ZCOOKIES*sizeof(u32),+GFP_KERNEL);+if(!skb)+return-ENOMEM;+rm->data.op_mmp_znotifier=RDS_ZCOPY_SKB(skb);+memset(rm->data.op_mmp_znotifier,0,+sizeof(*rm->data.op_mmp_znotifier));+if(mm_account_pinned_pages(&rm->data.op_mmp_znotifier->z_mmp,+length)){+consume_skb(skb);+rm->data.op_mmp_znotifier=NULL;+return-ENOMEM;+}+while(iov_iter_count(from)){+structpage*pages;+size_tstart;+ssize_tcopied;++copied=iov_iter_get_pages(from,&pages,PAGE_SIZE,+1,&start);+if(copied<0){+structmmpin*mmp;++mmp=&rm->data.op_mmp_znotifier->z_mmp;+mm_unaccount_pinned_pages(mmp);+consume_skb(skb);+rm->data.op_mmp_znotifier=NULL;+return-EFAULT;
also need to unmap pages pinned during previous iterations.
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-28 14:00:47
On Wed, Jan 24, 2018 at 12:46 PM, Sowmini Varadhan
[off-list ref] wrote:
quoted hunk
Send a cookie with sendmsg() on PF_RDS sockets, and process the
returned batched cookies in do_recv_completion()
Signed-off-by: Sowmini Varadhan <redacted>
---
tools/testing/selftests/net/msg_zerocopy.c | 119 ++++++++++++++++++++-------
1 files changed, 88 insertions(+), 31 deletions(-)
@@ -168,7 +168,26 @@ static int do_accept(int fd)returnfd;}-staticbooldo_sendmsg(intfd,structmsghdr*msg,booldo_zerocopy)+staticvoidadd_zcopy_cookie(structmsghdr*msg)+{+intolen=msg->msg_controllen;+structcmsghdr*cm;+staticuint32_tcookie;++msg->msg_controllen+=CMSG_SPACE(sizeof(cookie));+msg->msg_control=(structcmsghdr*)realloc(msg->msg_control,+msg->msg_controllen);
Please just allocate ahead of time. And since cookie size is fixed
and small just define a local variable on the stack in do_sendmsg.
char control[CMSG_SPACE(sizeof(uint32_t)];
+ if (!msg->msg_control)
+ error(1, errno, "cannot allocate cmsghdr for cookie");
+ cm = (void *)msg->msg_control + olen;
+ cm->cmsg_len = CMSG_SPACE(sizeof(cookie));
@@ -346,36 +382,57 @@ static bool do_recv_completion(int fd) cm->cmsg_level, cm->cmsg_type); serr = (void *) CMSG_DATA(cm);- if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY)- error(1, 0, "serr: wrong origin: %u", serr->ee_origin);- if (serr->ee_errno != 0)- error(1, 0, "serr: wrong error code: %u", serr->ee_errno);- hi = serr->ee_data;- lo = serr->ee_info;- range = hi - lo + 1;+ switch (serr->ee_origin) {+ case SO_EE_ORIGIN_ZEROCOPY: {+ if (serr->ee_errno != 0)+ error(1, 0, "serr: wrong error code: %u",+ serr->ee_errno);+ hi = serr->ee_data;+ lo = serr->ee_info;+ range = hi - lo + 1;++ /* Detect notification gaps. These should not happen often,+ * if at all. Gaps can occur due to drops, reordering and+ * retransmissions.+ */+ if (lo != next_completion)+ fprintf(stderr, "gap: %u..%u does not append to %u\n",+ lo, hi, next_completion);+ next_completion = hi + 1;++ zerocopy = !(serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED);+ if (zerocopied == -1)+ zerocopied = zerocopy;+ else if (zerocopied != zerocopy) {+ fprintf(stderr, "serr: inconsistent\n");+ zerocopied = zerocopy;+ }+ if (cfg_verbose >= 2)+ fprintf(stderr, "completed: %u (h=%u l=%u)\n",+ range, hi, lo);- /* Detect notification gaps. These should not happen often, if at all.- * Gaps can occur due to drops, reordering and retransmissions.- */- if (lo != next_completion)- fprintf(stderr, "gap: %u..%u does not append to %u\n",- lo, hi, next_completion);- next_completion = hi + 1;-- zerocopy = !(serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED);- if (zerocopied == -1)- zerocopied = zerocopy;- else if (zerocopied != zerocopy) {- fprintf(stderr, "serr: inconsistent\n");- zerocopied = zerocopy;
Instead of indenting all this existing code please add a helper
do_recv_completion_zcookie, call that if SO_EE_ORIGIN_ZCOOKIE
and fall through to existing code otherwise.
thanks for taking the time to go through the code!
An alternative that does not require a timer is to batch on the sk
error queue itself, like tcp zerocopy. That queues the first notification
skb on the error queue without any notification latency.
Then, if a subsequent notification comes in while another is pending
with < MAX zcookies, it coalesces the new notification onto the pending
skb and consumes the other. For RDS notifications, the implementation
is an extra skb_put + uint32_t assignment.
This is an interesting idea, let me give it a try.
Optionally, the socket can trigger another sk_error_report on each
new notification.
I was trying to avoid that- an upcall for each message is not a good
idea when you have high traffic and are able to free multiple rds_messages
at a time.
quoted
+static void rds_rm_zerocopy_callback(struct rds_sock *rs,
+ struct rds_znotifier *znotifier,
+ bool force)
+{
+ struct sock *sk = rds_rs_to_sk(rs);
+ struct sk_buff *skb;
+ struct sock_exterr_skb *serr;
+ unsigned long flags;
+ u32 *ptr;
+ int ncookies = 0, i;
+ struct rds_znotifier *znotif, *ztmp, *first;
+ LIST_HEAD(tmp_list);
+
+ spin_lock_irqsave(&rs->rs_lock, flags);
+ ncookies = rs->rs_ncookies;
+ if (ncookies < SO_EE_ORIGIN_MAX_ZCOOKIES && !force) {
+ if (znotifier) { /* add this cookie to the list and return */
can be checked before taking lock.
More importantly, when is this ever NULL?
It is null when invoked from tthe timer callback (rs_zcopy_notify()
going off because havent had any traffic for the expiration interval
so we want to send out pending notifications, but dont have any znotifier
in this case). But you are right in that:
This function is a callback
for a zerocopy struct of type znotifier. Is it doing double duty to flush
any outstanding if znotifier == NULL && force == true? If so, the first
condition probably never occurs unless force == true and thus the
second is redundant.
yes, force can simply be !znotifier.
I dont quite follow the "can be checked before taking the lock
comment though"- the lock is needed to make sure we atomically do
the lists "add new entry and potentially flush" operation.
the check is for the "potentially" part of that operation, so
I'm not seeing how it would help to move it out of the lock.
having said all that, I like the earlier suggestion of just
batching on the error_queue itself. If that works out without any
issues, all of this stuff may not be needed, so let me give that
a shot first.
This adds unnecessary notification latency to delivery of current
notification. The latest notification can be appended to tmp_list and
sent up immediately.
Resetting timeout on each queued notification causes unbound
notification latency for previous notifications on the queue.
I'm not sure I get that comment. RDS_REAP_TIMEOUT is the upper bound
for sending notification. If, in the interim, we get back a TCP
ack that lets us reap a bunch of messages, we'd go and flush the
queue anyway, so the RDS_REAP_TIMEOUT will not matter. Can you
elaborate on your concern?
You may alos be able to do the same as tcp zerocopy and
hold an sk reference on the notification skb.
;
I don't quite follow. Every notification skb is created when pages refcount
is increased. It persists until at least rds_rm_zerocopy_callback, after data
skb has been freed and pages refcount has been decreased.
In this callback, skb is consumed if another skb is already queued on
the error queue, otherwise it is queued itself. It needs to hold a sock ref
until it can be queued.
maybe I did not follow the original suggestion- were you
suggesting that I hold a pointer to the sk from e.g., the skb->cb
itself? I dont know that it would make things simpler,
whereas having the pointer and refcount in the rds_message itself,
and track this independantly of whether/not zcopy was used, seems
like a more consistent dsta-structure model, so I'd like to leave
this as is.
it's a static uint32_t in the function. It will get initialized to 0.
But the whole test program is rather simplistic, since it doesnt
actually verify the value of the cookies (hopefully me pending
updates to rds-stress will provide better testing/examples in this
space, because I actully have something multi-threaded, that must
necesarily use a truly random value for the cookie, and must really
make sure that the returned value from the kernel matches some
cookie that is pending)
quoted
+static void add_zcopy_cookie(struct msghdr *msg)
Please just allocate ahead of time. And since cookie size is fixed
and small just define a local variable on the stack in do_sendmsg.
Ok! I was just trying to make this as future-proof as possible, and
provide useful example code for the next cut/paste developer to use.
quoted
+ cm->cmsg_len = CMSG_SPACE(sizeof(cookie));
CMSG_LEN
:
Instead of indenting all this existing code please add a helper
do_recv_completion_zcookie, call that if SO_EE_ORIGIN_ZCOOKIE
and fall through to existing code otherwise.
:
Verify ncookies <= MAX_..
Verify ret == ncookies * sizeof(uint32_t)
quoted
+ zerocopied = 1;
Unused in this path
Ok, will fix all these.
Also, coalescing some related comments for patch 6/7:
Not read, so no need to configure. In that case a simpler
recv will do and is more concise than setting up recvmsg.
Real RDS applications actually have to use recvmsg, since they
can get cmsg info about other things like congestion notification
so let's leave this as recvmsg - it does no harm, and provides
test coverage for the recvmsg case as well.
it's a static uint32_t in the function. It will get initialized to 0.
Oh right. I missed that.
But the whole test program is rather simplistic, since it doesnt
actually verify the value of the cookies (hopefully me pending
updates to rds-stress will provide better testing/examples in this
space, because I actully have something multi-threaded, that must
necesarily use a truly random value for the cookie, and must really
make sure that the returned value from the kernel matches some
cookie that is pending)
It might be nice to at least increment the variable on each
successful send. The test is single threaded anyway. And
then we can test that the returned values are in the defined
range.
Also, coalescing some related comments for patch 6/7:
Not read, so no need to configure. In that case a simpler
recv will do and is more concise than setting up recvmsg.
Real RDS applications actually have to use recvmsg, since they
can get cmsg info about other things like congestion notification
so let's leave this as recvmsg - it does no harm, and provides
test coverage for the recvmsg case as well.
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-28 18:46:49
On Sun, Jan 28, 2018 at 5:15 PM, Sowmini Varadhan
[off-list ref] wrote:
thanks for taking the time to go through the code!
quoted
An alternative that does not require a timer is to batch on the sk
error queue itself, like tcp zerocopy. That queues the first notification
skb on the error queue without any notification latency.
Then, if a subsequent notification comes in while another is pending
with < MAX zcookies, it coalesces the new notification onto the pending
skb and consumes the other. For RDS notifications, the implementation
is an extra skb_put + uint32_t assignment.
This is an interesting idea, let me give it a try.
quoted
Optionally, the socket can trigger another sk_error_report on each
new notification.
I was trying to avoid that- an upcall for each message is not a good
idea when you have high traffic and are able to free multiple rds_messages
at a time.
Agreed. It was only a suggestion if that would be a reason
for you to not try the above idea.
quoted
quoted
+static void rds_rm_zerocopy_callback(struct rds_sock *rs,
+ struct rds_znotifier *znotifier,
+ bool force)
+{
+ struct sock *sk = rds_rs_to_sk(rs);
+ struct sk_buff *skb;
+ struct sock_exterr_skb *serr;
+ unsigned long flags;
+ u32 *ptr;
+ int ncookies = 0, i;
+ struct rds_znotifier *znotif, *ztmp, *first;
+ LIST_HEAD(tmp_list);
+
+ spin_lock_irqsave(&rs->rs_lock, flags);
+ ncookies = rs->rs_ncookies;
+ if (ncookies < SO_EE_ORIGIN_MAX_ZCOOKIES && !force) {
+ if (znotifier) { /* add this cookie to the list and return */
can be checked before taking lock.
More importantly, when is this ever NULL?
It is null when invoked from tthe timer callback (rs_zcopy_notify()
going off because havent had any traffic for the expiration interval
so we want to send out pending notifications, but dont have any znotifier
in this case). But you are right in that:
quoted
This function is a callback
for a zerocopy struct of type znotifier. Is it doing double duty to flush
any outstanding if znotifier == NULL && force == true? If so, the first
condition probably never occurs unless force == true and thus the
second is redundant.
yes, force can simply be !znotifier.
I dont quite follow the "can be checked before taking the lock
comment though"- the lock is needed to make sure we atomically do
the lists "add new entry and potentially flush" operation.
the check is for the "potentially" part of that operation, so
I'm not seeing how it would help to move it out of the lock.
having said all that, I like the earlier suggestion of just
batching on the error_queue itself. If that works out without any
issues, all of this stuff may not be needed, so let me give that
a shot first.
Sounds great!
quoted
This adds unnecessary notification latency to delivery of current
notification. The latest notification can be appended to tmp_list and
sent up immediately.
Resetting timeout on each queued notification causes unbound
notification latency for previous notifications on the queue.
I'm not sure I get that comment. RDS_REAP_TIMEOUT is the upper bound
for sending notification. If, in the interim, we get back a TCP
ack that lets us reap a bunch of messages, we'd go and flush the
queue anyway, so the RDS_REAP_TIMEOUT will not matter. Can you
elaborate on your concern?
I meant that if packet rate is low enough to require the timer
to fire, then by resetting the timer on each notification, the maximum
timeout for the first enqueued notification is the max cookie limit
* timeout, as opposed to timeout if set once and never modified.
Some physical device drivers do the same for their interrupt
moderation. You really want the configured timeout to be the
upper bound.
From: Willem de Bruijn <willemdebruijn.kernel@gmail.com> Date: 2018-01-28 18:55:39
On Sun, Jan 28, 2018 at 5:18 PM, Sowmini Varadhan
[off-list ref] wrote:
On (01/28/18 14:51), Willem de Bruijn wrote:
quoted
quoted
On (01/25/18 15:44), Willem de Bruijn wrote:
;
quoted
quoted
quoted
You may alos be able to do the same as tcp zerocopy and
hold an sk reference on the notification skb.
;
quoted
I don't quite follow. Every notification skb is created when pages refcount
is increased. It persists until at least rds_rm_zerocopy_callback, after data
skb has been freed and pages refcount has been decreased.
In this callback, skb is consumed if another skb is already queued on
the error queue, otherwise it is queued itself. It needs to hold a sock ref
until it can be queued.
maybe I did not follow the original suggestion- were you
suggesting that I hold a pointer to the sk from e.g., the skb->cb
itself?
Yes, I mean associating the notification skb that is eventually
queued onto the error queue with the socket. For tcp zerocopy,
this happens implicitly in sock_omalloc.
I dont know that it would make things simpler,
whereas having the pointer and refcount in the rds_message itself,
and track this independantly of whether/not zcopy was used, seems
like a more consistent dsta-structure model, so I'd like to leave
this as is.
Sounds good. You know the rds internals a lot better than I do,
so are the better judge on whether to choose that or sock_omalloc.
But the whole test program is rather simplistic, since it doesnt
actually verify the value of the cookies (hopefully me pending
:
It might be nice to at least increment the variable on each
successful send. The test is single threaded anyway. And
then we can test that the returned values are in the defined
range.
Yeah one thought that went through my head was that since
I now pick the cookes as consecutive numbers (1..N) anyway,
we could do a very simple-minded checksum on recv_completion
notification to check that the sum of all the cookie values
returned is actually N * (N+1)/2 - and flag errors/warnings
if it is not..
this would be a test enhancement I could do later.. first let
me try to avoid having all this complex timer-triggered code.
--Sowmini