tjh.dev / kernel
Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
kernel os linux
1
fork

Configure Feed

Select the types of activity you want to include in your feed.

sunrpc: convert queue_lock from global spinlock to per-cache-detail lock

The global queue_lock serializes all upcall queue operations across
every cache_detail instance. Convert it to a per-cache-detail spinlock
so that different caches (e.g. auth.unix.ip vs nfsd.fh) no longer
contend with each other on queue operations.

Signed-off-by: Jeff Layton <jlayton@kernel.org>
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>

authored by

Jeff Layton and committed by
Chuck Lever 17c1d665 6b4f16a5

+24 -24
+1
include/linux/sunrpc/cache.h
··· 113 113 114 114 /* fields for communication over channel */ 115 115 struct list_head queue; 116 + spinlock_t queue_lock; 116 117 117 118 atomic_t writers; /* how many time is /channel open */ 118 119 time64_t last_close; /* if no writers, when did last close */
+23 -24
net/sunrpc/cache.c
··· 400 400 { 401 401 spin_lock_init(&cd->hash_lock); 402 402 INIT_LIST_HEAD(&cd->queue); 403 + spin_lock_init(&cd->queue_lock); 403 404 spin_lock(&cache_list_lock); 404 405 cd->nextcheck = 0; 405 406 cd->entries = 0; ··· 804 803 * 805 804 */ 806 805 807 - static DEFINE_SPINLOCK(queue_lock); 808 - 809 806 struct cache_queue { 810 807 struct list_head list; 811 808 int reader; /* if 0, then request */ ··· 846 847 inode_lock(inode); /* protect against multiple concurrent 847 848 * readers on this file */ 848 849 again: 849 - spin_lock(&queue_lock); 850 + spin_lock(&cd->queue_lock); 850 851 /* need to find next request */ 851 852 while (rp->q.list.next != &cd->queue && 852 853 list_entry(rp->q.list.next, struct cache_queue, list) ··· 855 856 list_move(&rp->q.list, next); 856 857 } 857 858 if (rp->q.list.next == &cd->queue) { 858 - spin_unlock(&queue_lock); 859 + spin_unlock(&cd->queue_lock); 859 860 inode_unlock(inode); 860 861 WARN_ON_ONCE(rp->offset); 861 862 return 0; ··· 864 865 WARN_ON_ONCE(rq->q.reader); 865 866 if (rp->offset == 0) 866 867 rq->readers++; 867 - spin_unlock(&queue_lock); 868 + spin_unlock(&cd->queue_lock); 868 869 869 870 if (rq->len == 0) { 870 871 err = cache_request(cd, rq); ··· 875 876 876 877 if (rp->offset == 0 && !test_bit(CACHE_PENDING, &rq->item->flags)) { 877 878 err = -EAGAIN; 878 - spin_lock(&queue_lock); 879 + spin_lock(&cd->queue_lock); 879 880 list_move(&rp->q.list, &rq->q.list); 880 - spin_unlock(&queue_lock); 881 + spin_unlock(&cd->queue_lock); 881 882 } else { 882 883 if (rp->offset + count > rq->len) 883 884 count = rq->len - rp->offset; ··· 887 888 rp->offset += count; 888 889 if (rp->offset >= rq->len) { 889 890 rp->offset = 0; 890 - spin_lock(&queue_lock); 891 + spin_lock(&cd->queue_lock); 891 892 list_move(&rp->q.list, &rq->q.list); 892 - spin_unlock(&queue_lock); 893 + spin_unlock(&cd->queue_lock); 893 894 } 894 895 err = 0; 895 896 } 896 897 out: 897 898 if (rp->offset == 0) { 898 899 /* need to release rq */ 899 - spin_lock(&queue_lock); 900 + spin_lock(&cd->queue_lock); 900 901 rq->readers--; 901 902 if (rq->readers == 0 && 902 903 !test_bit(CACHE_PENDING, &rq->item->flags)) { 903 904 list_del(&rq->q.list); 904 - spin_unlock(&queue_lock); 905 + spin_unlock(&cd->queue_lock); 905 906 cache_put(rq->item, cd); 906 907 kfree(rq->buf); 907 908 kfree(rq); 908 909 } else 909 - spin_unlock(&queue_lock); 910 + spin_unlock(&cd->queue_lock); 910 911 } 911 912 if (err == -EAGAIN) 912 913 goto again; ··· 987 988 if (!rp) 988 989 return mask; 989 990 990 - spin_lock(&queue_lock); 991 + spin_lock(&cd->queue_lock); 991 992 992 993 for (cq= &rp->q; &cq->list != &cd->queue; 993 994 cq = list_entry(cq->list.next, struct cache_queue, list)) ··· 995 996 mask |= EPOLLIN | EPOLLRDNORM; 996 997 break; 997 998 } 998 - spin_unlock(&queue_lock); 999 + spin_unlock(&cd->queue_lock); 999 1000 return mask; 1000 1001 } 1001 1002 ··· 1010 1011 if (cmd != FIONREAD || !rp) 1011 1012 return -EINVAL; 1012 1013 1013 - spin_lock(&queue_lock); 1014 + spin_lock(&cd->queue_lock); 1014 1015 1015 1016 /* only find the length remaining in current request, 1016 1017 * or the length of the next request ··· 1023 1024 len = cr->len - rp->offset; 1024 1025 break; 1025 1026 } 1026 - spin_unlock(&queue_lock); 1027 + spin_unlock(&cd->queue_lock); 1027 1028 1028 1029 return put_user(len, (int __user *)arg); 1029 1030 } ··· 1045 1046 rp->offset = 0; 1046 1047 rp->q.reader = 1; 1047 1048 1048 - spin_lock(&queue_lock); 1049 + spin_lock(&cd->queue_lock); 1049 1050 list_add(&rp->q.list, &cd->queue); 1050 - spin_unlock(&queue_lock); 1051 + spin_unlock(&cd->queue_lock); 1051 1052 } 1052 1053 if (filp->f_mode & FMODE_WRITE) 1053 1054 atomic_inc(&cd->writers); ··· 1063 1064 if (rp) { 1064 1065 struct cache_request *rq = NULL; 1065 1066 1066 - spin_lock(&queue_lock); 1067 + spin_lock(&cd->queue_lock); 1067 1068 if (rp->offset) { 1068 1069 struct cache_queue *cq; 1069 1070 for (cq = &rp->q; &cq->list != &cd->queue; ··· 1085 1086 rp->offset = 0; 1086 1087 } 1087 1088 list_del(&rp->q.list); 1088 - spin_unlock(&queue_lock); 1089 + spin_unlock(&cd->queue_lock); 1089 1090 1090 1091 if (rq) { 1091 1092 cache_put(rq->item, cd); ··· 1112 1113 struct cache_request *cr; 1113 1114 LIST_HEAD(dequeued); 1114 1115 1115 - spin_lock(&queue_lock); 1116 + spin_lock(&detail->queue_lock); 1116 1117 list_for_each_entry_safe(cq, tmp, &detail->queue, list) 1117 1118 if (!cq->reader) { 1118 1119 cr = container_of(cq, struct cache_request, q); ··· 1125 1126 continue; 1126 1127 list_move(&cr->q.list, &dequeued); 1127 1128 } 1128 - spin_unlock(&queue_lock); 1129 + spin_unlock(&detail->queue_lock); 1129 1130 while (!list_empty(&dequeued)) { 1130 1131 cr = list_entry(dequeued.next, struct cache_request, q.list); 1131 1132 list_del(&cr->q.list); ··· 1250 1251 crq->buf = buf; 1251 1252 crq->len = 0; 1252 1253 crq->readers = 0; 1253 - spin_lock(&queue_lock); 1254 + spin_lock(&detail->queue_lock); 1254 1255 if (test_bit(CACHE_PENDING, &h->flags)) { 1255 1256 crq->item = cache_get(h); 1256 1257 list_add_tail(&crq->q.list, &detail->queue); ··· 1258 1259 } else 1259 1260 /* Lost a race, no longer PENDING, so don't enqueue */ 1260 1261 ret = -EAGAIN; 1261 - spin_unlock(&queue_lock); 1262 + spin_unlock(&detail->queue_lock); 1262 1263 wake_up(&queue_wait); 1263 1264 if (ret == -EAGAIN) { 1264 1265 kfree(buf);