1 /* 2 * Copyright (c) 2004-2016 Maxim Sobolev <sobomax@FreeBSD.org> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions 7 * are met: 8 * 1. Redistributions of source code must retain the above copyright 9 * notice, this list of conditions and the following disclaimer. 10 * 2. Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * 14 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND 15 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 16 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 17 * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE 18 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 19 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 20 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 21 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 22 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 23 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 24 * SUCH DAMAGE. 25 */ 26 27 #include <sys/cdefs.h> 28 __FBSDID("$FreeBSD$"); 29 30 #include <err.h> 31 #include <pthread.h> 32 #include <stdint.h> 33 #include <stdlib.h> 34 35 #if defined(MKUZ_DEBUG) 36 # include <assert.h> 37 #endif 38 39 #include "mkuzip.h" 40 #include "mkuz_fqueue.h" 41 #include "mkuz_conveyor.h" 42 #include "mkuz_blk.h" 43 #include "mkuz_blk_chain.h" 44 45 struct mkuz_fifo_queue * 46 mkuz_fqueue_ctor(int wakeup_len) 47 { 48 struct mkuz_fifo_queue *fqp; 49 50 fqp = mkuz_safe_zmalloc(sizeof(struct mkuz_fifo_queue)); 51 fqp->wakeup_len = wakeup_len; 52 if (pthread_mutex_init(&fqp->mtx, NULL) != 0) { 53 errx(1, "pthread_mutex_init() failed"); 54 } 55 if (pthread_cond_init(&fqp->cvar, NULL) != 0) { 56 errx(1, "pthread_cond_init() failed"); 57 } 58 return (fqp); 59 } 60 61 void 62 mkuz_fqueue_enq(struct mkuz_fifo_queue *fqp, struct mkuz_blk *bp) 63 { 64 struct mkuz_bchain_link *ip; 65 66 ip = mkuz_safe_zmalloc(sizeof(struct mkuz_bchain_link)); 67 ip->this = bp; 68 69 pthread_mutex_lock(&fqp->mtx); 70 if (fqp->first != NULL) { 71 fqp->first->prev = ip; 72 } else { 73 fqp->last = ip; 74 } 75 fqp->first = ip; 76 fqp->length += 1; 77 if (fqp->length >= fqp->wakeup_len) { 78 pthread_cond_signal(&fqp->cvar); 79 } 80 pthread_mutex_unlock(&fqp->mtx); 81 } 82 83 #if defined(NOTYET) 84 int 85 mkuz_fqueue_enq_all(struct mkuz_fifo_queue *fqp, struct mkuz_bchain_link *cip_f, 86 struct mkuz_bchain_link *cip_l, int clen) 87 { 88 int rval; 89 90 pthread_mutex_lock(&fqp->mtx); 91 if (fqp->first != NULL) { 92 fqp->first->prev = cip_l; 93 } else { 94 fqp->last = cip_l; 95 } 96 fqp->first = cip_f; 97 fqp->length += clen; 98 rval = fqp->length; 99 if (fqp->length >= fqp->wakeup_len) { 100 pthread_cond_signal(&fqp->cvar); 101 } 102 pthread_mutex_unlock(&fqp->mtx); 103 return (rval); 104 } 105 #endif 106 107 static int 108 mkuz_fqueue_check(struct mkuz_fifo_queue *fqp, cmp_cb_t cmp_cb, void *cap) 109 { 110 struct mkuz_bchain_link *ip; 111 112 for (ip = fqp->last; ip != NULL; ip = ip->prev) { 113 if (cmp_cb(ip->this, cap)) { 114 return (1); 115 } 116 } 117 return (0); 118 } 119 120 struct mkuz_blk * 121 mkuz_fqueue_deq_when(struct mkuz_fifo_queue *fqp, cmp_cb_t cmp_cb, void *cap) 122 { 123 struct mkuz_bchain_link *ip, *newlast, *newfirst, *mip; 124 struct mkuz_blk *bp; 125 126 pthread_mutex_lock(&fqp->mtx); 127 while (fqp->last == NULL || !mkuz_fqueue_check(fqp, cmp_cb, cap)) { 128 pthread_cond_wait(&fqp->cvar, &fqp->mtx); 129 } 130 if (cmp_cb(fqp->last->this, cap)) { 131 mip = fqp->last; 132 fqp->last = mip->prev; 133 if (fqp->last == NULL) { 134 #if defined(MKUZ_DEBUG) 135 assert(fqp->length == 1); 136 #endif 137 fqp->first = NULL; 138 } 139 } else { 140 #if defined(MKUZ_DEBUG) 141 assert(fqp->length > 1); 142 #endif 143 newfirst = newlast = fqp->last; 144 mip = NULL; 145 for (ip = fqp->last->prev; ip != NULL; ip = ip->prev) { 146 if (cmp_cb(ip->this, cap)) { 147 mip = ip; 148 continue; 149 } 150 newfirst->prev = ip; 151 newfirst = ip; 152 } 153 newfirst->prev = NULL; 154 fqp->first = newfirst; 155 fqp->last = newlast; 156 } 157 fqp->length -= 1; 158 pthread_mutex_unlock(&fqp->mtx); 159 bp = mip->this; 160 free(mip); 161 162 return bp; 163 } 164 165 struct mkuz_blk * 166 mkuz_fqueue_deq(struct mkuz_fifo_queue *fqp) 167 { 168 struct mkuz_bchain_link *ip; 169 struct mkuz_blk *bp; 170 171 pthread_mutex_lock(&fqp->mtx); 172 while (fqp->last == NULL) { 173 pthread_cond_wait(&fqp->cvar, &fqp->mtx); 174 } 175 #if defined(MKUZ_DEBUG) 176 assert(fqp->length > 0); 177 #endif 178 ip = fqp->last; 179 fqp->last = ip->prev; 180 if (fqp->last == NULL) { 181 #if defined(MKUZ_DEBUG) 182 assert(fqp->length == 1); 183 #endif 184 fqp->first = NULL; 185 } 186 fqp->length -= 1; 187 pthread_mutex_unlock(&fqp->mtx); 188 bp = ip->this; 189 free(ip); 190 191 return bp; 192 } 193 194 #if defined(NOTYET) 195 struct mkuz_bchain_link * 196 mkuz_fqueue_deq_all(struct mkuz_fifo_queue *fqp, int *rclen) 197 { 198 struct mkuz_bchain_link *rchain; 199 200 pthread_mutex_lock(&fqp->mtx); 201 while (fqp->last == NULL) { 202 pthread_cond_wait(&fqp->cvar, &fqp->mtx); 203 } 204 #if defined(MKUZ_DEBUG) 205 assert(fqp->length > 0); 206 #endif 207 rchain = fqp->last; 208 fqp->first = fqp->last = NULL; 209 *rclen = fqp->length; 210 fqp->length = 0; 211 pthread_mutex_unlock(&fqp->mtx); 212 return (rchain); 213 } 214 #endif 215