librsync  2.0.2
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  | The hard, lifeless I covered up the
24  | warm, pulsing It; protecting and
25  | sheltering.
26  */
27 
28 /** \file job.c Generic state-machine interface.
29  *
30  * The point of this is that we need to be able to suspend and resume
31  * processing at any point at which the buffers may block.
32  *
33  * \sa \ref api_streaming \sa rs_job_iter() \sa ::rs_job */
34 
35 #include "config.h"
36 
37 #include <stdlib.h>
38 #include <assert.h>
39 #include <stdio.h>
40 #include <time.h>
41 
42 #include "librsync.h"
43 #include "stream.h"
44 #include "util.h"
45 #include "sumset.h"
46 #include "job.h"
47 #include "trace.h"
48 
49 static rs_result rs_job_work(rs_job_t *job, rs_buffers_t *buffers);
50 
51 rs_job_t *rs_job_new(char const *job_name, rs_result (*statefn) (rs_job_t *))
52 {
53  rs_job_t *job;
54 
55  job = rs_alloc_struct(rs_job_t);
56 
57  job->job_name = job_name;
58  job->dogtag = RS_JOB_TAG;
59  job->statefn = statefn;
60 
61  job->stats.op = job_name;
62  job->stats.start = time(NULL);
63 
64  rs_trace("start %s job", job_name);
65 
66  return job;
67 }
68 
70 {
71  free(job->scoop_buf);
72  if (job->job_owns_sig)
74  rs_bzero(job, sizeof *job);
75  free(job);
76 
77  return RS_DONE;
78 }
79 
80 static rs_result rs_job_complete(rs_job_t *job, rs_result result)
81 {
82  rs_job_check(job);
83  assert(result != RS_RUNNING && result != RS_BLOCKED);
84  assert(rs_tube_is_idle(job) || result != RS_DONE);
85 
86  job->final_result = result;
87  job->stats.end = time(NULL);
88  if (result != RS_DONE) {
89  rs_error("%s job failed: %s", job->job_name, rs_strerror(result));
90  } else {
91  rs_trace("%s job complete", job->job_name);
92  }
93  return result;
94 }
95 
97 {
98  rs_result result;
99  size_t orig_in, orig_out;
100 
101  rs_job_check(job);
102  assert(buffers);
103 
104  orig_in = buffers->avail_in;
105  orig_out = buffers->avail_out;
106  result = rs_job_work(job, buffers);
107  if (result == RS_BLOCKED || result == RS_DONE)
108  if ((orig_in == buffers->avail_in) && (orig_out == buffers->avail_out)
109  && orig_in && orig_out) {
110  rs_error("internal error: job made no progress " "[orig_in="
111  FMT_SIZE ", orig_out=" FMT_SIZE ", final_in=" FMT_SIZE
112  ", final_out=" FMT_SIZE "]", orig_in, orig_out,
113  buffers->avail_in, buffers->avail_out);
114  return RS_INTERNAL_ERROR;
115  }
116  return result;
117 }
118 
119 static rs_result rs_job_work(rs_job_t *job, rs_buffers_t *buffers)
120 {
121  rs_result result;
122 
123  rs_job_check(job);
124  assert(buffers);
125 
126  job->stream = buffers;
127  while (1) {
128  result = rs_tube_catchup(job);
129  if (result == RS_DONE && job->statefn) {
130  result = job->statefn(job);
131  if (result == RS_DONE) {
132  /* The job is done so clear statefn. */
133  job->statefn = NULL;
134  /* There might be stuff in the tube, so keep running. */
135  continue;
136  }
137  }
138  if (result == RS_BLOCKED)
139  return result;
140  if (result != RS_RUNNING)
141  return rs_job_complete(job, result);
142  }
143 }
144 
146 {
147  return &job->stats;
148 }
149 
150 int rs_job_input_is_ending(rs_job_t *job)
151 {
152  return job->stream->eof_in;
153 }
154 
156  void *in_opaque, rs_driven_cb out_cb, void *out_opaque)
157 {
158  rs_result result, iores;
159 
160  rs_bzero(buf, sizeof *buf);
161 
162  do {
163  if (!buf->eof_in && in_cb) {
164  iores = in_cb(job, buf, in_opaque);
165  if (iores != RS_DONE)
166  return iores;
167  }
168 
169  result = rs_job_iter(job, buf);
170  if (result != RS_DONE && result != RS_BLOCKED)
171  return result;
172 
173  if (out_cb) {
174  iores = (out_cb) (job, buf, out_opaque);
175  if (iores != RS_DONE)
176  return iores;
177  }
178  } while (result != RS_DONE);
179 
180  return result;
181 }
Description of input and output buffers.
Definition: librsync.h:300
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:96
logging functions.
rs_result final_result
Final result of processing job.
Definition: job.h:38
rs_signature_t * signature
Pointer to the signature that's being used by the operation.
Definition: job.h:51
int job_owns_sig
Flag indicating signature should be destroyed with the job.
Definition: job.h:54
size_t avail_out
Remaining free space at next_out.
Definition: librsync.h:329
size_t avail_in
Number of bytes available at next_in.
Definition: librsync.h:314
void rs_free_sumset(rs_signature_t *)
Deep deallocation of checksums.
Definition: sumset.c:241
rs_result(* statefn)(rs_job_t *)
Callback for each processing step.
Definition: job.h:35
rs_stats_t stats
Encoding statistics.
Definition: job.h:72
Public header for librsync.
const char * job_name
Human-readable job operation name.
Definition: job.h:30
char const * op
Human-readable name of current operation.
Definition: librsync.h:192
Performance statistics from a librsync encoding or decoding operation.
Definition: librsync.h:191
char const * rs_strerror(rs_result r)
Return an English description of a rs_result value.
Definition: msg.c:49
rs_result
Return codes from nonblocking rsync operations.
Definition: librsync.h:161
int rs_tube_catchup(rs_job_t *job)
Put whatever will fit from the tube into the output of the stream.
Definition: tube.c:155
Blocked waiting for more data.
Definition: librsync.h:163
Probably a library bug.
Definition: librsync.h:180
const rs_stats_t * rs_job_statistics(rs_job_t *job)
Return a pointer to the statistics in a job.
Definition: job.c:145
rs_byte_t * scoop_buf
Buffer of data in the scoop.
Definition: job.h:77
rs_result rs_job_free(rs_job_t *job)
Deallocate job state.
Definition: job.c:69
The job is still running, and not yet finished or blocked.
Definition: librsync.h:164
int eof_in
True if there is no more data after this.
Definition: librsync.h:317
Completed successfully.
Definition: librsync.h:162
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:155
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:368
of this structure are private.
Definition: job.h:26