]> git.sur5r.net Git - bacula/bacula/blob - bacula/src/stored/append.c
Update doc
[bacula/bacula] / bacula / src / stored / append.c
1 /*
2  * Append code for Storage daemon
3  *  Kern Sibbald, May MM
4  *
5  *  Version $Id$
6  */
7 /*
8    Copyright (C) 2000-2003 Kern Sibbald and John Walker
9
10    This program is free software; you can redistribute it and/or
11    modify it under the terms of the GNU General Public License as
12    published by the Free Software Foundation; either version 2 of
13    the License, or (at your option) any later version.
14
15    This program is distributed in the hope that it will be useful,
16    but WITHOUT ANY WARRANTY; without even the implied warranty of
17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
18    General Public License for more details.
19
20    You should have received a copy of the GNU General Public
21    License along with this program; if not, write to the Free
22    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
23    MA 02111-1307, USA.
24
25  */
26
27 #include "bacula.h"
28 #include "stored.h"
29
30
31 /* Responses sent to the File daemon */
32 static char OK_data[]    = "3000 OK data\n";
33
34 /* Forward referenced functions */
35
36 /* 
37  *  Append Data sent from File daemon   
38  *
39  */
40 int do_append_data(JCR *jcr) 
41 {
42    int32_t n;
43    int32_t file_index, stream, last_file_index;
44    BSOCK *ds;
45    BSOCK *fd_sock = jcr->file_bsock;
46    bool ok = true;
47    DEVICE *dev;
48    DEV_RECORD rec;
49    DEV_BLOCK  *block;
50    DCR *dcr;
51    
52    Dmsg0(10, "Start append data.\n");
53
54    ds = fd_sock;
55
56    if (!bnet_set_buffer_size(ds, jcr->device->max_network_buffer_size, BNET_SETBUF_WRITE)) {
57       set_jcr_job_status(jcr, JS_ErrorTerminated);
58       Jmsg(jcr, M_FATAL, 0, _("Unable to set network buffer size.\n"));
59       return 0;
60    }
61
62    /* 
63     * Acquire output device for writing.  Note, after acquiring a
64     *   device, we MUST release it, which is done at the end of this
65     *   subroutine.
66     */
67    Dmsg0(100, "just before acquire_device\n");
68    if (!(dcr=acquire_device_for_append(jcr))) {
69       set_jcr_job_status(jcr, JS_ErrorTerminated);
70       return 0;
71    }
72    dev = dcr->dev;
73    block = dcr->block;
74
75    Dmsg1(20, "Begin append device=%s\n", dev_name(dev));
76
77    begin_data_spool(jcr);
78    begin_attribute_spool(jcr);
79
80    Dmsg0(100, "Just after acquire_device_for_append\n");
81    /*
82     * Write Begin Session Record
83     */
84    if (!write_session_label(jcr, block, SOS_LABEL)) {
85       Jmsg1(jcr, M_FATAL, 0, _("Write session label failed. ERR=%s\n"),
86          strerror_dev(dev));
87       set_jcr_job_status(jcr, JS_ErrorTerminated);
88       ok = false;
89    }
90
91    /* Tell File daemon to send data */
92    bnet_fsend(fd_sock, OK_data);
93
94    /* 
95     * Get Data from File daemon, write to device.  To clarify what is
96     *   going on here.  We expect:        
97     *     - A stream header
98     *     - Multiple records of data
99     *     - EOD record
100     *
101     *    The Stream header is just used to sychronize things, and
102     *    none of the stream header is written to tape.
103     *    The Multiple records of data, contain first the Attributes,
104     *    then after another stream header, the file data, then
105     *    after another stream header, the MD5 data if any.  
106     *
107     *   So we get the (stream header, data, EOD) three time for each
108     *   file. 1. for the Attributes, 2. for the file data if any, 
109     *   and 3. for the MD5 if any.
110     */
111    dcr->VolFirstIndex = dcr->VolLastIndex = 0;
112    jcr->run_time = time(NULL);              /* start counting time for rates */
113    for (last_file_index = 0; ok && !job_canceled(jcr); ) {
114
115       /* Read Stream header from the File daemon.
116        *  The stream header consists of the following:
117        *    file_index (sequential Bacula file index, base 1)
118        *    stream     (Bacula number to distinguish parts of data)
119        *    info       (Info for Storage daemon -- compressed, encryped, ...)
120        *       info is not currently used, so is read, but ignored!
121        */
122      if ((n=bget_msg(ds)) <= 0) {
123          if (n == BNET_SIGNAL && ds->msglen == BNET_EOD) {
124             break;                    /* end of data */
125          }
126          Jmsg1(jcr, M_FATAL, 0, _("Error reading data header from FD. ERR=%s\n"),
127             bnet_strerror(ds));
128          ok = false;
129          break;
130       }
131         
132       /* 
133        * This hand scanning is a bit more complicated than a simple
134        *   sscanf, but it allows us to handle any size integer up to
135        *   int64_t without worrying about whether %d, %ld, %lld, or %q 
136        *   is the correct format for each different architecture.
137        * It is a real pity that sscanf() is not portable.
138        */
139       char *p = ds->msg;
140       while (B_ISSPACE(*p)) {
141          p++;
142       }
143       file_index = (int32_t)str_to_int64(p);
144       while (B_ISDIGIT(*p)) {
145          p++;
146       }
147       if (!B_ISSPACE(*p) || !B_ISDIGIT(*(p+1))) {
148          Jmsg1(jcr, M_FATAL, 0, _("Malformed data header from FD: %s\n"), ds->msg);
149          ok = false;
150          break;
151       }
152       stream = (int32_t)str_to_int64(p);
153
154       Dmsg2(190, "<filed: Header FilInx=%d stream=%d\n", file_index, stream);
155
156       if (!(file_index > 0 && (file_index == last_file_index ||
157           file_index == last_file_index + 1))) {
158          Jmsg0(jcr, M_FATAL, 0, _("File index from FD not positive or sequential\n"));
159          ok = false;
160          break;
161       }
162       if (file_index != last_file_index) {
163          jcr->JobFiles = file_index;
164          last_file_index = file_index;
165       }
166       
167       /* Read data stream from the File daemon.
168        *  The data stream is just raw bytes
169        */
170       while ((n=bget_msg(ds)) > 0 && !job_canceled(jcr)) {
171          rec.VolSessionId = jcr->VolSessionId;
172          rec.VolSessionTime = jcr->VolSessionTime;
173          rec.FileIndex = file_index;
174          rec.Stream = stream;
175          rec.data_len = ds->msglen;
176          rec.data = ds->msg;            /* use message buffer */
177
178          Dmsg4(250, "before writ_rec FI=%d SessId=%d Strm=%s len=%d\n",
179             rec.FileIndex, rec.VolSessionId, stream_to_ascii(rec.Stream,rec.FileIndex), 
180             rec.data_len);
181           
182          while (!write_record_to_block(block, &rec)) {
183             Dmsg2(150, "!write_record_to_block data_len=%d rem=%d\n", rec.data_len,
184                        rec.remainder);
185             if (!write_block_to_device(jcr->dcr, block)) {
186                Dmsg2(90, "Got write_block_to_dev error on device %s. %s\n",
187                   dev_name(dev), strerror_dev(dev));
188                Jmsg(jcr, M_FATAL, 0, _("Fatal device error: ERR=%s\n"),
189                      strerror_dev(dev));
190                ok = false;
191                break;
192             }
193          }
194          if (!ok) {
195             Dmsg0(400, "Not OK\n");
196             break;
197          }
198          jcr->JobBytes += rec.data_len;   /* increment bytes this job */
199          Dmsg4(200, "write_record FI=%s SessId=%d Strm=%s len=%d\n",
200             FI_to_ascii(rec.FileIndex), rec.VolSessionId, 
201             stream_to_ascii(rec.Stream, rec.FileIndex), rec.data_len);
202
203          /* Send attributes and MD5 to Director for Catalog */
204          if (stream == STREAM_UNIX_ATTRIBUTES    || stream == STREAM_MD5_SIGNATURE ||
205              stream == STREAM_UNIX_ATTRIBUTES_EX || stream == STREAM_SHA1_SIGNATURE) { 
206             if (!jcr->no_attributes) {
207                if (are_attributes_spooled(jcr)) {
208                   jcr->dir_bsock->spool = true;
209                }
210                Dmsg0(200, "Send attributes.\n");
211                if (!dir_update_file_attributes(jcr, &rec)) {
212                   jcr->dir_bsock->spool = false;
213                   Jmsg(jcr, M_FATAL, 0, _("Error updating file attributes. ERR=%s\n"),
214                      bnet_strerror(jcr->dir_bsock));
215                   ok = false;
216                   break;
217                }
218                jcr->dir_bsock->spool = false;
219             }
220          }
221       }
222       if (is_bnet_error(ds)) {
223          Jmsg1(jcr, M_FATAL, 0, _("Network error on data channel. ERR=%s\n"),
224             bnet_strerror(ds));
225          ok = false;
226          break;
227       }
228    }
229
230    /* Create Job status for end of session label */
231    set_jcr_job_status(jcr, ok?JS_Terminated:JS_ErrorTerminated);
232
233    Dmsg1(200, "Write session label JobStatus=%d\n", jcr->JobStatus);
234
235    /*
236     * If !OK, check if we can still write. This may not be the case
237     *  if we are at the end of the tape or we got a fatal I/O error.
238     */
239    if (ok || dev_can_write(dev)) {
240       if (!write_session_label(jcr, block, EOS_LABEL)) {
241          Jmsg1(jcr, M_FATAL, 0, _("Error writting end session label. ERR=%s\n"),
242              strerror_dev(dev));
243          set_jcr_job_status(jcr, JS_ErrorTerminated);
244          ok = false;
245       }
246       Dmsg0(90, "back from write_end_session_label()\n");
247       /* Flush out final partial block of this session */
248       if (!write_block_to_device(jcr->dcr, block)) {
249          Dmsg0(100, _("Set ok=FALSE after write_block_to_device.\n"));
250          set_jcr_job_status(jcr, JS_ErrorTerminated);
251          ok = false;
252       }
253    }
254
255    if (!ok) {
256       discard_data_spool(jcr);
257    } else {
258       commit_data_spool(jcr);
259    }
260
261    Dmsg1(200, "calling release device JobStatus=%d\n", jcr->JobStatus);
262    /* Release the device */
263    if (!release_device(jcr)) {
264       Pmsg0(000, _("Error in release_device\n"));
265       set_jcr_job_status(jcr, JS_ErrorTerminated);
266       ok = false;
267    }
268
269    if (!ok) {
270       discard_attribute_spool(jcr);
271    } else {
272       commit_attribute_spool(jcr);
273    }
274
275    dir_send_job_status(jcr);          /* update director */
276
277    Dmsg1(100, "return from do_append_data() stat=%d\n", ok);
278    return ok ? 1 : 0;
279 }