librsync  2.0.1
job.c
Go to the documentation of this file.
1 /*= -*- c-basic-offset: 4; indent-tabs-mode: nil; -*-
2  *
3  * librsync -- the library for network deltas
4  *
5  * Copyright (C) 2000, 2001 by Martin Pool <mbp@sourcefrog.net>
6  *
7  * This program is free software; you can redistribute it and/or
8  * modify it under the terms of the GNU Lesser General Public License
9  * as published by the Free Software Foundation; either version 2.1 of
10  * the License, or (at your option) any later version.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15  * Lesser General Public License for more details.
16  *
17  * You should have received a copy of the GNU Lesser General Public
18  * License along with this program; if not, write to the Free Software
19  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
20  */
21 
22 
23  /*
24  | The hard, lifeless I covered up the
25  | warm, pulsing It; protecting and
26  | sheltering.
27  */
28 
29 /**
30  * \file job.c
31  *
32  * \brief
33  * Generic state-machine interface.
34  *
35  * The point of this is
36  * that we need to be able to suspend and resume processing at any
37  * point at which the buffers may block.
38  *
39  * \see \ref api_streaming
40  * \see rs_job_iter()
41  * \see ::rs_job
42  */
43 
44 
45 #include "config.h"
46 
47 #include <stdlib.h>
48 #include <assert.h>
49 #include <stdio.h>
50 #include <time.h>
51 
52 #include "librsync.h"
53 #include "stream.h"
54 #include "util.h"
55 #include "sumset.h"
56 #include "job.h"
57 #include "trace.h"
58 
59 
60 static const int rs_job_tag = 20010225;
61 
62 static rs_result rs_job_work(rs_job_t *job, rs_buffers_t *buffers);
63 
64 
65 rs_job_t * rs_job_new(char const *job_name, rs_result (*statefn)(rs_job_t *))
66 {
67  rs_job_t *job;
68 
69  job = rs_alloc_struct(rs_job_t);
70 
71  job->job_name = job_name;
72  job->dogtag = rs_job_tag;
73  job->statefn = statefn;
74 
75  job->stats.op = job_name;
76  job->stats.start = time(NULL);
77 
78  rs_trace("start %s job", job_name);
79 
80  return job;
81 }
82 
83 
84 void rs_job_check(rs_job_t *job)
85 {
86  assert(job->dogtag == rs_job_tag);
87 }
88 
89 
91 {
92  free(job->scoop_buf);
93  if (job->job_owns_sig)
95  rs_bzero(job, sizeof *job);
96  free(job);
97 
98  return RS_DONE;
99 }
100 
101 
102 
103 static rs_result rs_job_s_complete(rs_job_t *job)
104 {
105  rs_fatal("should not be reached");
106  return RS_INTERNAL_ERROR;
107 }
108 
109 
110 static rs_result rs_job_complete(rs_job_t *job, rs_result result)
111 {
112  rs_job_check(job);
113 
114  job->statefn = rs_job_s_complete;
115  job->final_result = result;
116 
117  if (result != RS_DONE) {
118  rs_error("%s job failed: %s", job->job_name, rs_strerror(result));
119  } else {
120  rs_trace("%s job complete", job->job_name);
121  }
122 
123  job->stats.end = time(NULL);
124  if (result == RS_DONE && !rs_tube_is_idle(job))
125  /* Processing is finished, but there is still some data
126  * waiting to get into the output buffer. */
127  return RS_BLOCKED;
128  else
129  return result;
130 }
131 
132 
134 {
135  rs_result result;
136  rs_long_t orig_in, orig_out;
137 
138  orig_in = buffers->avail_in;
139  orig_out = buffers->avail_out;
140 
141  result = rs_job_work(job, buffers);
142 
143  if (result == RS_BLOCKED || result == RS_DONE)
144  if ((orig_in == buffers->avail_in) && (orig_out == buffers->avail_out)
145  && orig_in && orig_out) {
146  rs_log(RS_LOG_ERR, "internal error: job made no progress "
147  "[orig_in=" PRINTF_FORMAT_U64 ", orig_out=" PRINTF_FORMAT_U64 ", final_in=" PRINTF_FORMAT_U64 ", final_out=" PRINTF_FORMAT_U64 "]",
148  PRINTF_CAST_U64(orig_in), PRINTF_CAST_U64(orig_out), PRINTF_CAST_U64(buffers->avail_in),
149  PRINTF_CAST_U64(buffers->avail_out));
150  return RS_INTERNAL_ERROR;
151  }
152 
153  return result;
154 }
155 
156 
157 static rs_result
158 rs_job_work(rs_job_t *job, rs_buffers_t *buffers)
159 {
160  rs_result result;
161 
162  rs_job_check(job);
163 
164  if (!buffers) {
165  rs_error("NULL buffer passed to rs_job_iter");
166  return RS_PARAM_ERROR;
167  }
168  job->stream = buffers;
169 
170  while (1) {
171  result = rs_tube_catchup(job);
172  if (result == RS_BLOCKED)
173  return result;
174  else if (result != RS_DONE)
175  return rs_job_complete(job, result);
176 
177  if (job->statefn == rs_job_s_complete) {
178  if (rs_tube_is_idle(job))
179  return RS_DONE;
180  else
181  return RS_BLOCKED;
182  } else {
183  result = job->statefn(job);
184  if (result == RS_RUNNING)
185  continue;
186  else if (result == RS_BLOCKED)
187  return result;
188  else
189  return rs_job_complete(job, result);
190  }
191  }
192 
193  /* TODO: Before returning, check that we actually made some
194  * progress. If not, and we're not returning an error, this is a
195  * bug. */
196 }
197 
198 
199 const rs_stats_t *
201 {
202  return &job->stats;
203 }
204 
205 
206 int
207 rs_job_input_is_ending(rs_job_t *job)
208 {
209  return job->stream->eof_in;
210 }
211 
212 
213 
214 rs_result
216  rs_driven_cb in_cb, void *in_opaque,
217  rs_driven_cb out_cb, void *out_opaque)
218 {
219  rs_result result, iores;
220 
221  rs_bzero(buf, sizeof *buf);
222 
223  do {
224  if (!buf->eof_in && in_cb) {
225  iores = in_cb(job, buf, in_opaque);
226  if (iores != RS_DONE)
227  return iores;
228  }
229 
230  result = rs_job_iter(job, buf);
231  if (result != RS_DONE && result != RS_BLOCKED)
232  return result;
233 
234  if (out_cb) {
235  iores = (out_cb)(job, buf, out_opaque);
236  if (iores != RS_DONE)
237  return iores;
238  }
239  } while (result != RS_DONE);
240 
241  return result;
242 }
Description of input and output buffers.
Definition: librsync.h:361
rs_result rs_job_iter(rs_job_t *job, rs_buffers_t *buffers)
Run a rs_job state machine until it blocks (RS_BLOCKED), returns an error, or completes (RS_DONE)...
Definition: job.c:133
Bad value passed in to library, probably an application bug.
Definition: librsync.h:217
rs_result final_result
Final result of processing job.
Definition: job.h:41
rs_signature_t * signature
Pointer to the signature that&#39;s being used by the operation.
Definition: job.h:54
int job_owns_sig
Flag indicating signature should be destroyed with the job.
Definition: job.h:57
size_t avail_out
Remaining free space at next_out.
Definition: librsync.h:397
long long rs_long_t
A long integer type that can handle the largest file offsets.
size_t avail_in
Number of bytes available at next_in References the length of available input.
Definition: librsync.h:376
void rs_free_sumset(rs_signature_t *)
Deep deallocation of checksums.
Definition: sumset.c:201
rs_result(* statefn)(rs_job_t *)
Callback for each processing step.
Definition: job.h:38
rs_stats_t stats
Encoding statistics.
Definition: job.h:75
Error conditions.
Definition: librsync.h:123
Public header for librsync.
const char * job_name
Human-readable job operation name.
Definition: job.h:33
char const * op
Human-readable name of current operation.
Definition: librsync.h:237
Performance statistics from a librsync encoding or decoding operation.
Definition: librsync.h:236
char const * rs_strerror(rs_result r)
Return an English description of a rs_result value.
Definition: msg.c:51
rs_result
Return codes from nonblocking rsync operations.
Definition: librsync.h:193
Blocked waiting for more data.
Definition: librsync.h:195
Probably a library bug.
Definition: librsync.h:216
const rs_stats_t * rs_job_statistics(rs_job_t *job)
Return a pointer to the statistics in a job.
Definition: job.c:200
rs_byte_t * scoop_buf
Buffer of data in the scoop.
Definition: job.h:82
rs_result rs_job_free(rs_job_t *job)
Deallocate job state.
Definition: job.c:90
The job is still running, and not yet finished or blocked.
Definition: librsync.h:199
int eof_in
True if there is no more data after this.
Definition: librsync.h:381
Completed successfully.
Definition: librsync.h:194
rs_result rs_job_drive(rs_job_t *job, rs_buffers_t *buf, rs_driven_cb in_cb, void *in_opaque, rs_driven_cb out_cb, void *out_opaque)
Actively process a job, by making callbacks to fill and empty the buffers until the job is done...
Definition: job.c:215
rs_result rs_driven_cb(rs_job_t *job, rs_buffers_t *buf, void *opaque)
Type of application-supplied function for rs_job_drive().
Definition: librsync.h:448
The contents of this structure are private.
Definition: job.h:29