$treeview $search $mathjax $extrastylesheet
librsync
2.0.2
$projectbrief
|
$projectbrief
|
$searchbox |
00001 /*= -*- c-basic-offset: 4; indent-tabs-mode: nil; -*- 00002 * 00003 * librsync -- the library for network deltas 00004 * 00005 * Copyright (C) 2000, 2001 by Martin Pool <mbp@sourcefrog.net> 00006 * 00007 * This program is free software; you can redistribute it and/or 00008 * modify it under the terms of the GNU Lesser General Public License 00009 * as published by the Free Software Foundation; either version 2.1 of 00010 * the License, or (at your option) any later version. 00011 * 00012 * This program is distributed in the hope that it will be useful, but 00013 * WITHOUT ANY WARRANTY; without even the implied warranty of 00014 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 00015 * Lesser General Public License for more details. 00016 * 00017 * You should have received a copy of the GNU Lesser General Public 00018 * License along with this program; if not, write to the Free Software 00019 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. 00020 */ 00021 00022 /*= 00023 | The hard, lifeless I covered up the 00024 | warm, pulsing It; protecting and 00025 | sheltering. 00026 */ 00027 00028 /** \file job.c Generic state-machine interface. 00029 * 00030 * The point of this is that we need to be able to suspend and resume 00031 * processing at any point at which the buffers may block. 00032 * 00033 * \sa \ref api_streaming \sa rs_job_iter() \sa ::rs_job */ 00034 00035 #include "config.h" 00036 00037 #include <stdlib.h> 00038 #include <assert.h> 00039 #include <stdio.h> 00040 #include <time.h> 00041 00042 #include "librsync.h" 00043 #include "stream.h" 00044 #include "util.h" 00045 #include "sumset.h" 00046 #include "job.h" 00047 #include "trace.h" 00048 00049 static rs_result rs_job_work(rs_job_t *job, rs_buffers_t *buffers); 00050 00051 rs_job_t *rs_job_new(char const *job_name, rs_result (*statefn) (rs_job_t *)) 00052 { 00053 rs_job_t *job; 00054 00055 job = rs_alloc_struct(rs_job_t); 00056 00057 job->job_name = job_name; 00058 job->dogtag = RS_JOB_TAG; 00059 job->statefn = statefn; 00060 00061 job->stats.op = job_name; 00062 job->stats.start = time(NULL); 00063 00064 rs_trace("start %s job", job_name); 00065 00066 return job; 00067 } 00068 00069 rs_result rs_job_free(rs_job_t *job) 00070 { 00071 free(job->scoop_buf); 00072 if (job->job_owns_sig) 00073 rs_free_sumset(job->signature); 00074 rs_bzero(job, sizeof *job); 00075 free(job); 00076 00077 return RS_DONE; 00078 } 00079 00080 static rs_result rs_job_complete(rs_job_t *job, rs_result result) 00081 { 00082 rs_job_check(job); 00083 assert(result != RS_RUNNING && result != RS_BLOCKED); 00084 assert(rs_tube_is_idle(job) || result != RS_DONE); 00085 00086 job->final_result = result; 00087 job->stats.end = time(NULL); 00088 if (result != RS_DONE) { 00089 rs_error("%s job failed: %s", job->job_name, rs_strerror(result)); 00090 } else { 00091 rs_trace("%s job complete", job->job_name); 00092 } 00093 return result; 00094 } 00095 00096 rs_result rs_job_iter(rs_job_t *job, rs_buffers_t *buffers) 00097 { 00098 rs_result result; 00099 size_t orig_in, orig_out; 00100 00101 rs_job_check(job); 00102 assert(buffers); 00103 00104 orig_in = buffers->avail_in; 00105 orig_out = buffers->avail_out; 00106 result = rs_job_work(job, buffers); 00107 if (result == RS_BLOCKED || result == RS_DONE) 00108 if ((orig_in == buffers->avail_in) && (orig_out == buffers->avail_out) 00109 && orig_in && orig_out) { 00110 rs_error("internal error: job made no progress " "[orig_in=" 00111 FMT_SIZE ", orig_out=" FMT_SIZE ", final_in=" FMT_SIZE 00112 ", final_out=" FMT_SIZE "]", orig_in, orig_out, 00113 buffers->avail_in, buffers->avail_out); 00114 return RS_INTERNAL_ERROR; 00115 } 00116 return result; 00117 } 00118 00119 static rs_result rs_job_work(rs_job_t *job, rs_buffers_t *buffers) 00120 { 00121 rs_result result; 00122 00123 rs_job_check(job); 00124 assert(buffers); 00125 00126 job->stream = buffers; 00127 while (1) { 00128 result = rs_tube_catchup(job); 00129 if (result == RS_DONE && job->statefn) { 00130 result = job->statefn(job); 00131 if (result == RS_DONE) { 00132 /* The job is done so clear statefn. */ 00133 job->statefn = NULL; 00134 /* There might be stuff in the tube, so keep running. */ 00135 continue; 00136 } 00137 } 00138 if (result == RS_BLOCKED) 00139 return result; 00140 if (result != RS_RUNNING) 00141 return rs_job_complete(job, result); 00142 } 00143 } 00144 00145 const rs_stats_t *rs_job_statistics(rs_job_t *job) 00146 { 00147 return &job->stats; 00148 } 00149 00150 int rs_job_input_is_ending(rs_job_t *job) 00151 { 00152 return job->stream->eof_in; 00153 } 00154 00155 rs_result rs_job_drive(rs_job_t *job, rs_buffers_t *buf, rs_driven_cb in_cb, 00156 void *in_opaque, rs_driven_cb out_cb, void *out_opaque) 00157 { 00158 rs_result result, iores; 00159 00160 rs_bzero(buf, sizeof *buf); 00161 00162 do { 00163 if (!buf->eof_in && in_cb) { 00164 iores = in_cb(job, buf, in_opaque); 00165 if (iores != RS_DONE) 00166 return iores; 00167 } 00168 00169 result = rs_job_iter(job, buf); 00170 if (result != RS_DONE && result != RS_BLOCKED) 00171 return result; 00172 00173 if (out_cb) { 00174 iores = (out_cb) (job, buf, out_opaque); 00175 if (iores != RS_DONE) 00176 return iores; 00177 } 00178 } while (result != RS_DONE); 00179 00180 return result; 00181 }