1 /* Background I/O service for Redis.
2 *
3 * This file implements operations that we need to perform in the background.
4 * Currently there is only a single operation, that is a background close(2)
5 * system call. This is needed as when the process is the last owner of a
6 * reference to a file closing it means unlinking it, and the deletion of the
7 * file is slow, blocking the server.
8 *
9 * In the future we'll either continue implementing new things we need or
10 * we'll switch to libeio. However there are probably long term uses for this
11 * file as we may want to put here Redis specific background tasks (for instance
12 * it is not impossible that we'll need a non blocking FLUSHDB/FLUSHALL
13 * implementation).
14 *
15 * DESIGN
16 * ------
17 *
18 * The design is trivial, we have a structure representing a job to perform
19 * and a different thread and job queue for every job type.
20 * Every thread wait for new jobs in its queue, and process every job
21 * sequentially.
22 *
23 * Jobs of the same type are guaranteed to be processed from the least
24 * recently inserted to the most recently inserted (older jobs processed
25 * first).
26 *
27 * Currently there is no way for the creator of the job to be notified about
28 * the completion of the operation, this will only be added when/if needed.
29 *
30 * ----------------------------------------------------------------------------
31 *
32 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
33 * All rights reserved.
34 *
35 * Redistribution and use in source and binary forms, with or without
36 * modification, are permitted provided that the following conditions are met:
37 *
38 * * Redistributions of source code must retain the above copyright notice,
39 * this list of conditions and the following disclaimer.
40 * * Redistributions in binary form must reproduce the above copyright
41 * notice, this list of conditions and the following disclaimer in the
42 * documentation and/or other materials provided with the distribution.
43 * * Neither the name of Redis nor the names of its contributors may be used
44 * to endorse or promote products derived from this software without
45 * specific prior written permission.
46 *
47 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
48 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
49 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
50 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
51 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
52 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
53 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
54 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
55 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
56 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
57 * POSSIBILITY OF SUCH DAMAGE.
58 */
59
60
61 #include "server.h"
62 #include "bio.h"
63
64 static pthread_t bio_threads[BIO_NUM_OPS];
65 static pthread_mutex_t bio_mutex[BIO_NUM_OPS];
66 static pthread_cond_t bio_condvar[BIO_NUM_OPS];
67 static list *bio_jobs[BIO_NUM_OPS];
68 /* The following array is used to hold the number of pending jobs for every
69 * OP type. This allows us to export the bioPendingJobsOfType() API that is
70 * useful when the main thread wants to perform some operation that may involve
71 * objects shared with the background thread. The main thread will just wait
72 * that there are no longer jobs of this type to be executed before performing
73 * the sensible operation. This data is also useful for reporting. */
74 static unsigned long long bio_pending[BIO_NUM_OPS];
75
76 /* This structure represents a background Job. It is only used locally to this
77 * file as the API does not expose the internals at all. */
78 struct bio_job {
79 time_t time; /* Time at which the job was created. */
80 /* Job specific arguments pointers. If we need to pass more than three
81 * arguments we can just pass a pointer to a structure or alike. */
82 void *arg1, *arg2, *arg3;
83 };
84
85 void *bioProcessBackgroundJobs(void *arg);
86
87 /* Make sure we have enough stack to perform all the things we do in the
88 * main thread. */
89 #define REDIS_THREAD_STACK_SIZE (1024*1024*4)
90
91 /* Initialize the background system, spawning the thread. */
bioInit(void)92 void bioInit(void) {
93 pthread_attr_t attr;
94 pthread_t thread;
95 size_t stacksize;
96 int j;
97
98 /* Initialization of state vars and objects */
99 for (j = 0; j < BIO_NUM_OPS; j++) {
100 pthread_mutex_init(&bio_mutex[j],NULL);
101 pthread_cond_init(&bio_condvar[j],NULL);
102 bio_jobs[j] = listCreate();
103 bio_pending[j] = 0;
104 }
105
106 /* Set the stack size as by default it may be small in some system */
107 pthread_attr_init(&attr);
108 pthread_attr_getstacksize(&attr,&stacksize);
109 if (!stacksize) stacksize = 1; /* The world is full of Solaris Fixes */
110 while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
111 pthread_attr_setstacksize(&attr, stacksize);
112
113 /* Ready to spawn our threads. We use the single argument the thread
114 * function accepts in order to pass the job ID the thread is
115 * responsible of. */
116 for (j = 0; j < BIO_NUM_OPS; j++) {
117 void *arg = (void*)(unsigned long) j;
118 if (pthread_create(&thread,&attr,bioProcessBackgroundJobs,arg) != 0) {
119 serverLog(LL_WARNING,"Fatal: Can't initialize Background Jobs.");
120 exit(1);
121 }
122 bio_threads[j] = thread;
123 }
124 }
125
bioCreateBackgroundJob(int type,void * arg1,void * arg2,void * arg3)126 void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3) {
127 struct bio_job *job = zmalloc(sizeof(*job));
128
129 job->time = time(NULL);
130 job->arg1 = arg1;
131 job->arg2 = arg2;
132 job->arg3 = arg3;
133 pthread_mutex_lock(&bio_mutex[type]);
134 listAddNodeTail(bio_jobs[type],job);
135 bio_pending[type]++;
136 pthread_cond_signal(&bio_condvar[type]);
137 pthread_mutex_unlock(&bio_mutex[type]);
138 }
139
bioProcessBackgroundJobs(void * arg)140 void *bioProcessBackgroundJobs(void *arg) {
141 struct bio_job *job;
142 unsigned long type = (unsigned long) arg;
143 sigset_t sigset;
144
145 /* Check that the type is within the right interval. */
146 if (type >= BIO_NUM_OPS) {
147 serverLog(LL_WARNING,
148 "Warning: bio thread started with wrong type %lu",type);
149 return NULL;
150 }
151
152 /* Make the thread killable at any time, so that bioKillThreads()
153 * can work reliably. */
154 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
155 pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
156
157 pthread_mutex_lock(&bio_mutex[type]);
158 /* Block SIGALRM so we are sure that only the main thread will
159 * receive the watchdog signal. */
160 sigemptyset(&sigset);
161 sigaddset(&sigset, SIGALRM);
162 if (pthread_sigmask(SIG_BLOCK, &sigset, NULL))
163 serverLog(LL_WARNING,
164 "Warning: can't mask SIGALRM in bio.c thread: %s", strerror(errno));
165
166 while(1) {
167 listNode *ln;
168
169 /* The loop always starts with the lock hold. */
170 if (listLength(bio_jobs[type]) == 0) {
171 pthread_cond_wait(&bio_condvar[type],&bio_mutex[type]);
172 continue;
173 }
174 /* Pop the job from the queue. */
175 ln = listFirst(bio_jobs[type]);
176 job = ln->value;
177 /* It is now possible to unlock the background system as we know have
178 * a stand alone job structure to process.*/
179 pthread_mutex_unlock(&bio_mutex[type]);
180
181 /* Process the job accordingly to its type. */
182 if (type == BIO_CLOSE_FILE) {
183 close((long)job->arg1);
184 } else if (type == BIO_AOF_FSYNC) {
185 aof_fsync((long)job->arg1);
186 } else {
187 serverPanic("Wrong job type in bioProcessBackgroundJobs().");
188 }
189 zfree(job);
190
191 /* Lock again before reiterating the loop, if there are no longer
192 * jobs to process we'll block again in pthread_cond_wait(). */
193 pthread_mutex_lock(&bio_mutex[type]);
194 listDelNode(bio_jobs[type],ln);
195 bio_pending[type]--;
196 }
197 }
198
199 /* Return the number of pending jobs of the specified type. */
bioPendingJobsOfType(int type)200 unsigned long long bioPendingJobsOfType(int type) {
201 unsigned long long val;
202 pthread_mutex_lock(&bio_mutex[type]);
203 val = bio_pending[type];
204 pthread_mutex_unlock(&bio_mutex[type]);
205 return val;
206 }
207
208 /* Kill the running bio threads in an unclean way. This function should be
209 * used only when it's critical to stop the threads for some reason.
210 * Currently Redis does this only on crash (for instance on SIGSEGV) in order
211 * to perform a fast memory check without other threads messing with memory. */
bioKillThreads(void)212 void bioKillThreads(void) {
213 int err, j;
214
215 for (j = 0; j < BIO_NUM_OPS; j++) {
216 if (pthread_cancel(bio_threads[j]) == 0) {
217 if ((err = pthread_join(bio_threads[j],NULL)) != 0) {
218 serverLog(LL_WARNING,
219 "Bio thread for job type #%d can be joined: %s",
220 j, strerror(err));
221 } else {
222 serverLog(LL_WARNING,
223 "Bio thread for job type #%d terminated",j);
224 }
225 }
226 }
227 }
228