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