3 * Bacula Director -- msgchan.c -- handles the message channel
4 * to the Storage daemon and the File daemon.
6 * Kern Sibbald, August MM
8 * This routine runs as a thread and must be thread reentrant.
10 * Basic tasks done here:
11 * Open a message channel with the Storage daemon
12 * to authenticate ourself and to pass the JobId.
13 * Create a thread to interact with the Storage daemon
14 * who returns a job status and requests Catalog services, etc.
19 Copyright (C) 2000, 2001, 2002 Kern Sibbald and John Walker
21 This program is free software; you can redistribute it and/or
22 modify it under the terms of the GNU General Public License as
23 published by the Free Software Foundation; either version 2 of
24 the License, or (at your option) any later version.
26 This program is distributed in the hope that it will be useful,
27 but WITHOUT ANY WARRANTY; without even the implied warranty of
28 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
29 General Public License for more details.
31 You should have received a copy of the GNU General Public
32 License along with this program; if not, write to the Free
33 Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
41 /* Commands sent to Storage daemon */
42 static char jobcmd[] = "JobId=%d job=%s job_name=%s client_name=%s \
43 type=%d level=%d FileSet=%s Allow=%s Session=%s\n";
44 static char use_device[] = "use device=%s media_type=%s pool_name=%s pool_type=%s\n";
46 /* Response from Storage daemon */
47 static char OKjob[] = "3000 OK Job SDid=%d SDtime=%d Authorization=%100s\n";
48 static char OK_device[] = "3000 OK use device\n";
50 /* Storage Daemon requests */
51 static char Job_start[] = "3010 Job %127s start\n";
52 static char Job_end[] =
53 "3099 Job %127s end JobStatus=%d JobFiles=%d JobBytes=%" lld "\n";
54 static char Job_status[] = "3012 Job %127s jobstatus %d\n";
56 /* Forward referenced functions */
57 static void *msg_thread(void *arg);
60 * Establish a message channel connection with the Storage daemon
61 * and perform authentication.
63 int connect_to_storage_daemon(JCR *jcr, int retry_interval,
64 int max_retry_time, int verbose)
69 * Open message channel with the Storage daemon
71 Dmsg2(200, "bnet_connect to Storage daemon %s:%d\n", jcr->store->address,
73 sd = bnet_connect(jcr, retry_interval, max_retry_time,
74 _("Storage daemon"), jcr->store->address,
75 NULL, jcr->store->SDport, verbose);
79 sd->res = (RES *)jcr->store; /* save pointer to other end */
80 jcr->store_bsock = sd;
82 if (!authenticate_storage_daemon(jcr)) {
89 * Start a job with the Storage daemon
91 int start_storage_daemon_job(JCR *jcr)
96 char auth_key[100]; /* max 17 chars */
97 char *device_name, *pool_name, *pool_type, *media_type;
98 int device_name_len, pool_name_len, pool_type_len, media_type_len;
100 storage = jcr->store;
101 sd = jcr->store_bsock;
103 * Now send JobId and permissions, and get back the authorization key.
105 bash_spaces(jcr->job->hdr.name);
106 bash_spaces(jcr->client->hdr.name);
107 bash_spaces(jcr->client->hdr.name);
108 bnet_fsend(sd, jobcmd, jcr->JobId, jcr->Job, jcr->job->hdr.name,
109 jcr->client->hdr.name, jcr->JobType, jcr->JobLevel,
110 jcr->fileset->hdr.name, "append", "*");
111 unbash_spaces(jcr->job->hdr.name);
112 unbash_spaces(jcr->client->hdr.name);
113 unbash_spaces(jcr->fileset->hdr.name);
114 if (bnet_recv(sd) > 0) {
115 Dmsg1(110, "<stored: %s", sd->msg);
116 if (sscanf(sd->msg, OKjob, &jcr->VolSessionId,
117 &jcr->VolSessionTime, &auth_key) != 3) {
118 Jmsg(jcr, M_FATAL, 0, _("Storage daemon rejected Job command: %s\n"), sd->msg);
121 jcr->sd_auth_key = bstrdup(auth_key);
122 Dmsg1(150, "sd_auth_key=%s\n", jcr->sd_auth_key);
125 Jmsg(jcr, M_FATAL, 0, _("<stored: bad response to Job command: %s\n"),
131 * Send use device = xxx media = yyy pool = zzz
133 device_name_len = strlen(storage->dev_name) + 1;
134 media_type_len = strlen(storage->media_type) + 1;
135 pool_type_len = strlen(jcr->pool->pool_type) + 1;
136 pool_name_len = strlen(jcr->pool->hdr.name) + 1;
137 device_name = (char *) get_memory(device_name_len);
138 pool_name = (char *) get_memory(pool_name_len);
139 pool_type = (char *) get_memory(pool_type_len);
140 media_type = (char *) get_memory(media_type_len);
141 memcpy(device_name, storage->dev_name, device_name_len);
142 memcpy(media_type, storage->media_type, media_type_len);
143 memcpy(pool_type, jcr->pool->pool_type, pool_type_len);
144 memcpy(pool_name, jcr->pool->hdr.name, pool_name_len);
145 bash_spaces(device_name);
146 bash_spaces(media_type);
147 bash_spaces(pool_type);
148 bash_spaces(pool_name);
149 sd->msg = (char *) check_pool_memory_size(sd->msg, sizeof(device_name) +
150 device_name_len + media_type_len + pool_type_len + pool_name_len);
151 bnet_fsend(sd, use_device, device_name, media_type, pool_name, pool_type);
152 Dmsg1(110, ">stored: %s", sd->msg);
153 status = response(sd, OK_device, "Use Device");
155 free_memory(device_name);
156 free_memory(media_type);
157 free_memory(pool_name);
158 free_memory(pool_type);
164 * Start a thread to handle Storage daemon messages and
167 int start_storage_daemon_message_thread(JCR *jcr)
173 jcr->use_count++; /* mark in use by msg thread */
175 if ((status=pthread_create(&thid, NULL, msg_thread, (void *)jcr)) != 0) {
176 Emsg1(M_ABORT, 0, _("Cannot create message thread: %s\n"), strerror(status));
178 jcr->SD_msg_chan = thid;
182 static void msg_thread_cleanup(void *arg)
184 JCR *jcr = (JCR *)arg;
185 Dmsg0(200, "End msg_thread\n");
186 db_end_transaction(jcr->db); /* terminate any open transaction */
188 jcr->msg_thread_done = TRUE;
189 pthread_cond_broadcast(&jcr->term_wait); /* wakeup any waiting threads */
191 free_jcr(jcr); /* release jcr */
195 * Handle the message channel (i.e. requests from the
197 * Note, we are running in a separate thread.
199 static void *msg_thread(void *arg)
201 JCR *jcr = (JCR *)arg;
204 char Job[MAX_NAME_LENGTH];
209 pthread_cleanup_push(msg_thread_cleanup, arg);
210 Dmsg0(200, "msg_thread\n");
211 sd = jcr->store_bsock;
212 pthread_detach(pthread_self());
214 /* Read the Storage daemon's output.
216 Dmsg0(200, "Start msg_thread loop\n");
217 while ((stat=bget_msg(sd, 0)) > 0) {
218 Dmsg1(200, "<stored: %s", sd->msg);
219 if (sscanf(sd->msg, Job_start, &Job) == 1) {
222 if (sscanf(sd->msg, Job_end, &Job, &JobStatus, &JobFiles,
224 jcr->SDJobStatus = JobStatus; /* termination status */
225 jcr->JobFiles = JobFiles;
226 jcr->JobBytes = JobBytes;
229 if (sscanf(sd->msg, Job_status, &Job, &JobStatus) == 2) {
230 jcr->SDJobStatus = JobStatus; /* current status */
235 jcr->SDJobStatus = JS_ErrorTerminated;
237 pthread_cleanup_pop(1);
241 void wait_for_storage_daemon_termination(JCR *jcr)
243 /* Now wait for Storage daemon to terminate our message thread */
245 jcr->JobStatus = JS_WaitSD;
246 while (!jcr->msg_thread_done && !job_cancelled(jcr)) {
249 struct timespec timeout;
251 gettimeofday(&tv, &tz);
253 timeout.tv_sec = tv.tv_sec + 10; /* wait 10 seconds */
254 Dmsg0(300, "I'm waiting for message thread termination.\n");
255 pthread_cond_timedwait(&jcr->term_wait, &jcr->mutex, &timeout);