sunrpc: split cache_detail queue into request and reader lists

+62 -85

2 changed files

expand all

include

linux

sunrpc

cache.h

net

sunrpc

cache.c

+3 -1

include/linux/sunrpc/cache.h

··· 113 113 int entries; 114 114 115 115 /* fields for communication over channel */ 116 - struct list_head queue; 116 + struct list_head requests; 117 + struct list_head readers; 117 118 spinlock_t queue_lock; 118 119 wait_queue_head_t queue_wait; 120 + u64 next_seqno; 119 121 120 122 atomic_t writers; /* how many time is /channel open */ 121 123 time64_t last_close; /* if no writers, when did last close */

+59 -84

net/sunrpc/cache.c

··· 399 399 void sunrpc_init_cache_detail(struct cache_detail *cd) 400 400 { 401 401 spin_lock_init(&cd->hash_lock); 402 - INIT_LIST_HEAD(&cd->queue); 402 + INIT_LIST_HEAD(&cd->requests); 403 + INIT_LIST_HEAD(&cd->readers); 403 404 spin_lock_init(&cd->queue_lock); 404 405 init_waitqueue_head(&cd->queue_wait); 406 + cd->next_seqno = 0; 405 407 spin_lock(&cache_list_lock); 406 408 cd->nextcheck = 0; 407 409 cd->entries = 0; ··· 798 796 * On read, you get a full request, or block. 799 797 * On write, an update request is processed. 800 798 * Poll works if anything to read, and always allows write. 801 - * 802 - * Implemented by linked list of requests. Each open file has 803 - * a ->private that also exists in this list. New requests are added 804 - * to the end and may wakeup and preceding readers. 805 - * New readers are added to the head. If, on read, an item is found with 806 - * CACHE_UPCALLING clear, we free it from the list. 807 - * 808 799 */ 809 800 810 - struct cache_queue { 811 - struct list_head list; 812 - int reader; /* if 0, then request */ 813 - }; 814 801 struct cache_request { 815 - struct cache_queue q; 802 + struct list_head list; 816 803 struct cache_head *item; 817 - char * buf; 804 + char *buf; 818 805 int len; 819 806 int readers; 807 + u64 seqno; 820 808 }; 821 809 struct cache_reader { 822 - struct cache_queue q; 810 + struct list_head list; 823 811 int offset; /* if non-0, we have a refcnt on next request */ 812 + u64 next_seqno; 824 813 }; 825 814 826 815 static int cache_request(struct cache_detail *detail, ··· 824 831 if (len < 0) 825 832 return -E2BIG; 826 833 return PAGE_SIZE - len; 834 + } 835 + 836 + static struct cache_request * 837 + cache_next_request(struct cache_detail *cd, u64 seqno) 838 + { 839 + struct cache_request *rq; 840 + 841 + list_for_each_entry(rq, &cd->requests, list) 842 + if (rq->seqno >= seqno) 843 + return rq; 844 + return NULL; 827 845 } 828 846 829 847 static ssize_t cache_read(struct file *filp, char __user *buf, size_t count, ··· 853 849 again: 854 850 spin_lock(&cd->queue_lock); 855 851 /* need to find next request */ 856 - while (rp->q.list.next != &cd->queue && 857 - list_entry(rp->q.list.next, struct cache_queue, list) 858 - ->reader) { 859 - struct list_head *next = rp->q.list.next; 860 - list_move(&rp->q.list, next); 861 - } 862 - if (rp->q.list.next == &cd->queue) { 852 + rq = cache_next_request(cd, rp->next_seqno); 853 + if (!rq) { 863 854 spin_unlock(&cd->queue_lock); 864 855 inode_unlock(inode); 865 856 WARN_ON_ONCE(rp->offset); 866 857 return 0; 867 858 } 868 - rq = container_of(rp->q.list.next, struct cache_request, q.list); 869 - WARN_ON_ONCE(rq->q.reader); 870 859 if (rp->offset == 0) 871 860 rq->readers++; 872 861 spin_unlock(&cd->queue_lock); ··· 873 876 874 877 if (rp->offset == 0 && !test_bit(CACHE_PENDING, &rq->item->flags)) { 875 878 err = -EAGAIN; 876 - spin_lock(&cd->queue_lock); 877 - list_move(&rp->q.list, &rq->q.list); 878 - spin_unlock(&cd->queue_lock); 879 + rp->next_seqno = rq->seqno + 1; 879 880 } else { 880 881 if (rp->offset + count > rq->len) 881 882 count = rq->len - rp->offset; ··· 883 888 rp->offset += count; 884 889 if (rp->offset >= rq->len) { 885 890 rp->offset = 0; 886 - spin_lock(&cd->queue_lock); 887 - list_move(&rp->q.list, &rq->q.list); 888 - spin_unlock(&cd->queue_lock); 891 + rp->next_seqno = rq->seqno + 1; 889 892 } 890 893 err = 0; 891 894 } ··· 894 901 rq->readers--; 895 902 if (rq->readers == 0 && 896 903 !test_bit(CACHE_PENDING, &rq->item->flags)) { 897 - list_del(&rq->q.list); 904 + list_del(&rq->list); 898 905 spin_unlock(&cd->queue_lock); 899 906 cache_put(rq->item, cd); 900 907 kfree(rq->buf); ··· 969 976 { 970 977 __poll_t mask; 971 978 struct cache_reader *rp = filp->private_data; 972 - struct cache_queue *cq; 973 979 974 980 poll_wait(filp, &cd->queue_wait, wait); 975 981 ··· 980 988 981 989 spin_lock(&cd->queue_lock); 982 990 983 - for (cq= &rp->q; &cq->list != &cd->queue; 984 - cq = list_entry(cq->list.next, struct cache_queue, list)) 985 - if (!cq->reader) { 986 - mask |= EPOLLIN | EPOLLRDNORM; 987 - break; 988 - } 991 + if (cache_next_request(cd, rp->next_seqno)) 992 + mask |= EPOLLIN | EPOLLRDNORM; 989 993 spin_unlock(&cd->queue_lock); 990 994 return mask; 991 995 } ··· 992 1004 { 993 1005 int len = 0; 994 1006 struct cache_reader *rp = filp->private_data; 995 - struct cache_queue *cq; 1007 + struct cache_request *rq; 996 1008 997 1009 if (cmd != FIONREAD || !rp) 998 1010 return -EINVAL; ··· 1002 1014 /* only find the length remaining in current request, 1003 1015 * or the length of the next request 1004 1016 */ 1005 - for (cq= &rp->q; &cq->list != &cd->queue; 1006 - cq = list_entry(cq->list.next, struct cache_queue, list)) 1007 - if (!cq->reader) { 1008 - struct cache_request *cr = 1009 - container_of(cq, struct cache_request, q); 1010 - len = cr->len - rp->offset; 1011 - break; 1012 - } 1017 + rq = cache_next_request(cd, rp->next_seqno); 1018 + if (rq) 1019 + len = rq->len - rp->offset; 1013 1020 spin_unlock(&cd->queue_lock); 1014 1021 1015 1022 return put_user(len, (int __user *)arg); ··· 1025 1042 return -ENOMEM; 1026 1043 } 1027 1044 rp->offset = 0; 1028 - rp->q.reader = 1; 1045 + rp->next_seqno = 0; 1029 1046 1030 1047 spin_lock(&cd->queue_lock); 1031 - list_add(&rp->q.list, &cd->queue); 1048 + list_add(&rp->list, &cd->readers); 1032 1049 spin_unlock(&cd->queue_lock); 1033 1050 } 1034 1051 if (filp->f_mode & FMODE_WRITE) ··· 1047 1064 1048 1065 spin_lock(&cd->queue_lock); 1049 1066 if (rp->offset) { 1050 - struct cache_queue *cq; 1051 - for (cq = &rp->q; &cq->list != &cd->queue; 1052 - cq = list_entry(cq->list.next, 1053 - struct cache_queue, list)) 1054 - if (!cq->reader) { 1055 - struct cache_request *cr = 1056 - container_of(cq, 1057 - struct cache_request, q); 1058 - cr->readers--; 1059 - if (cr->readers == 0 && 1060 - !test_bit(CACHE_PENDING, 1061 - &cr->item->flags)) { 1062 - list_del(&cr->q.list); 1063 - rq = cr; 1064 - } 1065 - break; 1067 + struct cache_request *cr; 1068 + 1069 + cr = cache_next_request(cd, rp->next_seqno); 1070 + if (cr) { 1071 + cr->readers--; 1072 + if (cr->readers == 0 && 1073 + !test_bit(CACHE_PENDING, 1074 + &cr->item->flags)) { 1075 + list_del(&cr->list); 1076 + rq = cr; 1066 1077 } 1078 + } 1067 1079 rp->offset = 0; 1068 1080 } 1069 - list_del(&rp->q.list); 1081 + list_del(&rp->list); 1070 1082 spin_unlock(&cd->queue_lock); 1071 1083 1072 1084 if (rq) { ··· 1085 1107 1086 1108 static void cache_dequeue(struct cache_detail *detail, struct cache_head *ch) 1087 1109 { 1088 - struct cache_queue *cq, *tmp; 1089 - struct cache_request *cr; 1110 + struct cache_request *cr, *tmp; 1090 1111 LIST_HEAD(dequeued); 1091 1112 1092 1113 spin_lock(&detail->queue_lock); 1093 - list_for_each_entry_safe(cq, tmp, &detail->queue, list) 1094 - if (!cq->reader) { 1095 - cr = container_of(cq, struct cache_request, q); 1096 - if (cr->item != ch) 1097 - continue; 1098 - if (test_bit(CACHE_PENDING, &ch->flags)) 1099 - /* Lost a race and it is pending again */ 1100 - break; 1101 - if (cr->readers != 0) 1102 - continue; 1103 - list_move(&cr->q.list, &dequeued); 1104 - } 1114 + list_for_each_entry_safe(cr, tmp, &detail->requests, list) { 1115 + if (cr->item != ch) 1116 + continue; 1117 + if (test_bit(CACHE_PENDING, &ch->flags)) 1118 + /* Lost a race and it is pending again */ 1119 + break; 1120 + if (cr->readers != 0) 1121 + continue; 1122 + list_move(&cr->list, &dequeued); 1123 + } 1105 1124 spin_unlock(&detail->queue_lock); 1106 1125 while (!list_empty(&dequeued)) { 1107 - cr = list_entry(dequeued.next, struct cache_request, q.list); 1108 - list_del(&cr->q.list); 1126 + cr = list_entry(dequeued.next, struct cache_request, list); 1127 + list_del(&cr->list); 1109 1128 cache_put(cr->item, detail); 1110 1129 kfree(cr->buf); 1111 1130 kfree(cr); ··· 1220 1245 return -EAGAIN; 1221 1246 } 1222 1247 1223 - crq->q.reader = 0; 1224 1248 crq->buf = buf; 1225 1249 crq->len = 0; 1226 1250 crq->readers = 0; 1227 1251 spin_lock(&detail->queue_lock); 1228 1252 if (test_bit(CACHE_PENDING, &h->flags)) { 1229 1253 crq->item = cache_get(h); 1230 - list_add_tail(&crq->q.list, &detail->queue); 1254 + crq->seqno = detail->next_seqno++; 1255 + list_add_tail(&crq->list, &detail->requests); 1231 1256 trace_cache_entry_upcall(detail, h); 1232 1257 } else 1233 1258 /* Lost a race, no longer PENDING, so don't enqueue */

Configure Feed

Configure Feed