git.gag.com Git - debian/amanda/blob - device-src/rait-device.c

   1 /*
   2  * Copyright (c) 2007-2012 Zmanda, Inc.  All Rights Reserved.
   3  *
   4  * This program is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU General Public License
   6  * as published by the Free Software Foundation; either version 2
   7  * of the License, or (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  11  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  12  * for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  17  *
  18  * Contact information: Zmanda Inc., 465 S. Mathilda Ave., Suite 300
  19  * Sunnyvale, CA 94085, USA, or: http://www.zmanda.com
  20  */
  21
  22 /* The RAIT device encapsulates some number of other devices into a single
  23  * redundant device. */
  24
  25 #include "amanda.h"
  26 #include "property.h"
  27 #include "util.h"
  28 #include <glib.h>
  29 #include "glib-util.h"
  30 #include "device.h"
  31 #include "fileheader.h"
  32 #include "amsemaphore.h"
  33
  34 /* Just a note about the failure mode of different operations:
  35    - Recovers from a failure (enters degraded mode)
  36      open_device()
  37      seek_file() -- explodes if headers don't match.
  38      seek_block() -- explodes if headers don't match.
  39      read_block() -- explodes if data doesn't match.
  40
  41    - Operates in degraded mode (but dies if a new problem shows up)
  42      read_label() -- but dies on label mismatch.
  43      start() -- but dies when writing in degraded mode.
  44      property functions
  45      finish()
  46
  47    - Dies in degraded mode (even if remaining devices are OK)
  48      start_file()
  49      write_block()
  50      finish_file()
  51      recycle_file()
  52 */
  53
  54 /*
  55  * Type checking and casting macros
  56  */
  57 #define TYPE_RAIT_DEVICE        (rait_device_get_type())
  58 #define RAIT_DEVICE(obj)        G_TYPE_CHECK_INSTANCE_CAST((obj), rait_device_get_type(), RaitDevice)
  59 #define RAIT_DEVICE_CONST(obj)  G_TYPE_CHECK_INSTANCE_CAST((obj), rait_device_get_type(), RaitDevice const)
  60 #define RAIT_DEVICE_CLASS(klass)        G_TYPE_CHECK_CLASS_CAST((klass), rait_device_get_type(), RaitDeviceClass)
  61 #define IS_RAIT_DEVICE(obj)     G_TYPE_CHECK_INSTANCE_TYPE((obj), rait_device_get_type ())
  62
  63 #define RAIT_DEVICE_GET_CLASS(obj)      G_TYPE_INSTANCE_GET_CLASS((obj), rait_device_get_type(), RaitDeviceClass)
  64 static GType    rait_device_get_type    (void);
  65
  66 /*
  67  * Main object structure
  68  */
  69 typedef struct RaitDevice_s {
  70     Device __parent__;
  71
  72     struct RaitDevicePrivate_s * private;
  73 } RaitDevice;
  74
  75 /*
  76  * Class definition
  77  */
  78 typedef struct _RaitDeviceClass RaitDeviceClass;
  79 struct _RaitDeviceClass {
  80     DeviceClass __parent__;
  81 };
  82
  83 typedef enum {
  84     RAIT_STATUS_COMPLETE, /* All subdevices OK. */
  85     RAIT_STATUS_DEGRADED, /* One subdevice failed. */
  86     RAIT_STATUS_FAILED    /* Two or more subdevices failed. */
  87 } RaitStatus;
  88
  89 /* Older versions of glib have a deadlock in their thread pool implementations,
  90  * so we include a simple thread-pool implementation here to replace it.
  91  *
  92  * This implementation assumes that threads are used for paralellizing a single
  93  * operation, so all threads run a function to completion before the main thread
  94  * continues.  This simplifies some of the locking semantics, and in particular
  95  * there is no need to wait for stray threads to finish an operation when
  96  * finalizing the RaitDevice object or when beginning a new operation.
  97  */
  98 #if !(GLIB_CHECK_VERSION(2,10,0))
  99 #define USE_INTERNAL_THREADPOOL
 100 #endif
 101
 102 typedef struct RaitDevicePrivate_s {
 103     GPtrArray * children;
 104     /* These flags are only relevant for reading. */
 105     RaitStatus status;
 106     /* If status == RAIT_STATUS_DEGRADED, this holds the index of the
 107        failed node. It holds a negative number otherwise. */
 108     int failed;
 109
 110     /* the child block size */
 111     gsize child_block_size;
 112
 113 #ifdef USE_INTERNAL_THREADPOOL
 114     /* array of ThreadInfo for performing parallel operations */
 115     GArray *threads;
 116
 117     /* value of this semaphore is the number of threaded operations
 118      * in progress */
 119     amsemaphore_t *threads_sem;
 120 #endif
 121 } RaitDevicePrivate;
 122
 123 #ifdef USE_INTERNAL_THREADPOOL
 124 typedef struct ThreadInfo {
 125     GThread *thread;
 126
 127     /* struct fields below are protected by this mutex and condition variable */
 128     GMutex *mutex;
 129     GCond *cond;
 130
 131     gboolean die;
 132     GFunc func;
 133     gpointer data;
 134
 135     /* give threads access to active_threads and its mutex/cond */
 136     struct RaitDevicePrivate_s *private;
 137 } ThreadInfo;
 138 #endif
 139
 140 /* This device uses a special sentinel node to indicate that the child devices
 141  * will be set later (in rait_device_open).  It contains a control character to
 142  * make it difficult to enter accidentally in an Amanda config. */
 143 #define DEFER_CHILDREN_SENTINEL "DEFER\1"
 144
 145 #define PRIVATE(o) (o->private)
 146
 147 #define rait_device_in_error(dev) \
 148     (device_in_error((dev)) || PRIVATE(RAIT_DEVICE((dev)))->status == RAIT_STATUS_FAILED)
 149
 150 void rait_device_register (void);
 151
 152 /* here are local prototypes */
 153 static void rait_device_init (RaitDevice * o);
 154 static void rait_device_class_init (RaitDeviceClass * c);
 155 static void rait_device_base_init (RaitDeviceClass * c);
 156 static void rait_device_open_device (Device * self, char * device_name, char * device_type, char * device_node);
 157 static gboolean rait_device_start (Device * self, DeviceAccessMode mode,
 158                                    char * label, char * timestamp);
 159 static gboolean rait_device_configure(Device * self, gboolean use_global_config);
 160 static gboolean rait_device_start_file(Device * self, dumpfile_t * info);
 161 static gboolean rait_device_write_block (Device * self, guint size, gpointer data);
 162 static gboolean rait_device_finish_file (Device * self);
 163 static dumpfile_t * rait_device_seek_file (Device * self, guint file);
 164 static gboolean rait_device_seek_block (Device * self, guint64 block);
 165 static int      rait_device_read_block (Device * self, gpointer buf,
 166                                         int * size);
 167 static gboolean rait_device_recycle_file (Device * self, guint filenum);
 168 static gboolean rait_device_finish (Device * self);
 169 static DeviceStatusFlags rait_device_read_label(Device * dself);
 170 static void find_simple_params(RaitDevice * self, guint * num_children,
 171                                guint * data_children);
 172
 173 /* property handlers */
 174
 175 static gboolean property_get_block_size_fn(Device *self,
 176     DevicePropertyBase *base, GValue *val,
 177     PropertySurety *surety, PropertySource *source);
 178
 179 static gboolean property_set_block_size_fn(Device *self,
 180     DevicePropertyBase *base, GValue *val,
 181     PropertySurety surety, PropertySource source);
 182
 183 static gboolean property_get_canonical_name_fn(Device *self,
 184     DevicePropertyBase *base, GValue *val,
 185     PropertySurety *surety, PropertySource *source);
 186
 187 static gboolean property_get_concurrency_fn(Device *self,
 188     DevicePropertyBase *base, GValue *val,
 189     PropertySurety *surety, PropertySource *source);
 190
 191 static gboolean property_get_streaming_fn(Device *self,
 192     DevicePropertyBase *base, GValue *val,
 193     PropertySurety *surety, PropertySource *source);
 194
 195 static gboolean property_get_boolean_and_fn(Device *self,
 196     DevicePropertyBase *base, GValue *val,
 197     PropertySurety *surety, PropertySource *source);
 198
 199 static gboolean property_get_medium_access_type_fn(Device *self,
 200     DevicePropertyBase *base, GValue *val,
 201     PropertySurety *surety, PropertySource *source);
 202
 203 static gboolean property_get_max_volume_usage_fn(Device *self,
 204     DevicePropertyBase *base, GValue *val,
 205     PropertySurety *surety, PropertySource *source);
 206
 207 static gboolean property_set_max_volume_usage_fn(Device *self,
 208     DevicePropertyBase *base, GValue *val,
 209     PropertySurety surety, PropertySource source);
 210
 211
 212 /* pointer to the class of our parent */
 213 static DeviceClass *parent_class = NULL;
 214
 215 static GType
 216 rait_device_get_type (void)
 217 {
 218     static GType type = 0;
 219
 220     if G_UNLIKELY(type == 0) {
 221         static const GTypeInfo info = {
 222             sizeof (RaitDeviceClass),
 223             (GBaseInitFunc) rait_device_base_init,
 224             (GBaseFinalizeFunc) NULL,
 225             (GClassInitFunc) rait_device_class_init,
 226             (GClassFinalizeFunc) NULL,
 227             NULL /* class_data */,
 228             sizeof (RaitDevice),
 229             0 /* n_preallocs */,
 230             (GInstanceInitFunc) rait_device_init,
 231             NULL
 232         };
 233
 234         type = g_type_register_static (TYPE_DEVICE, "RaitDevice", &info,
 235                                        (GTypeFlags)0);
 236         }
 237
 238     return type;
 239 }
 240
 241 static void g_object_unref_foreach(gpointer data,
 242                                    gpointer user_data G_GNUC_UNUSED) {
 243     if (data != NULL && G_IS_OBJECT(data)) {
 244         g_object_unref(data);
 245     }
 246 }
 247
 248 static void
 249 rait_device_finalize(GObject *obj_self)
 250 {
 251     RaitDevice *self = RAIT_DEVICE (obj_self);
 252     if(G_OBJECT_CLASS(parent_class)->finalize) \
 253            (* G_OBJECT_CLASS(parent_class)->finalize)(obj_self);
 254     if(self->private->children) {
 255         g_ptr_array_foreach(self->private->children,
 256                             g_object_unref_foreach, NULL);
 257         g_ptr_array_free (self->private->children, TRUE);
 258         self->private->children = NULL;
 259     }
 260 #ifdef USE_INTERNAL_THREADPOOL
 261     g_assert(PRIVATE(self)->threads_sem == NULL || PRIVATE(self)->threads_sem->value == 0);
 262
 263     if (PRIVATE(self)->threads) {
 264         guint i;
 265
 266         for (i = 0; i < PRIVATE(self)->threads->len; i++) {
 267             ThreadInfo *inf = &g_array_index(PRIVATE(self)->threads, ThreadInfo, i);
 268             if (inf->thread) {
 269                 /* NOTE: the thread is waiting on this condition right now, not
 270                  * executing an operation. */
 271
 272                 /* ask the thread to die */
 273                 g_mutex_lock(inf->mutex);
 274                 inf->die = TRUE;
 275                 g_cond_signal(inf->cond);
 276                 g_mutex_unlock(inf->mutex);
 277
 278                 /* and wait for it to die, which should happen soon */
 279                 g_thread_join(inf->thread);
 280             }
 281
 282             if (inf->mutex)
 283                 g_mutex_free(inf->mutex);
 284             if (inf->cond)
 285                 g_cond_free(inf->cond);
 286         }
 287     }
 288
 289     if (PRIVATE(self)->threads_sem)
 290         amsemaphore_free(PRIVATE(self)->threads_sem);
 291 #endif
 292     amfree(self->private);
 293 }
 294
 295 static void
 296 rait_device_init (RaitDevice * o G_GNUC_UNUSED)
 297 {
 298     PRIVATE(o) = g_new(RaitDevicePrivate, 1);
 299     PRIVATE(o)->children = g_ptr_array_new();
 300     PRIVATE(o)->status = RAIT_STATUS_COMPLETE;
 301     PRIVATE(o)->failed = -1;
 302 #ifdef USE_INTERNAL_THREADPOOL
 303     PRIVATE(o)->threads = NULL;
 304     PRIVATE(o)->threads_sem = NULL;
 305 #endif
 306 }
 307
 308 static void
 309 rait_device_class_init (RaitDeviceClass * c)
 310 {
 311     GObjectClass *g_object_class = (GObjectClass*) c;
 312     DeviceClass *device_class = (DeviceClass *)c;
 313
 314     parent_class = g_type_class_ref (TYPE_DEVICE);
 315
 316     device_class->open_device = rait_device_open_device;
 317     device_class->configure = rait_device_configure;
 318     device_class->start = rait_device_start;
 319     device_class->start_file = rait_device_start_file;
 320     device_class->write_block = rait_device_write_block;
 321     device_class->finish_file = rait_device_finish_file;
 322     device_class->seek_file = rait_device_seek_file;
 323     device_class->seek_block = rait_device_seek_block;
 324     device_class->read_block = rait_device_read_block;
 325     device_class->recycle_file = rait_device_recycle_file;
 326     device_class->finish = rait_device_finish;
 327     device_class->read_label = rait_device_read_label;
 328
 329     g_object_class->finalize = rait_device_finalize;
 330
 331 #ifndef USE_INTERNAL_THREADPOOL
 332 #if !GLIB_CHECK_VERSION(2,10,2)
 333     /* Versions of glib before 2.10.2 crash if
 334      * g_thread_pool_set_max_unused_threads is called before the first
 335      * invocation of g_thread_pool_new.  So we make up a thread pool, but don't
 336      * start any threads in it, and free it */
 337     {
 338         GThreadPool *pool = g_thread_pool_new((GFunc)-1, NULL, -1, FALSE, NULL);
 339         g_thread_pool_free(pool, TRUE, FALSE);
 340     }
 341 #endif
 342
 343     g_thread_pool_set_max_unused_threads(-1);
 344 #endif
 345 }
 346
 347 static void
 348 rait_device_base_init (RaitDeviceClass * c)
 349 {
 350     DeviceClass *device_class = (DeviceClass *)c;
 351
 352     /* the RAIT device overrides most of the standard properties, so that it
 353      * can calculate them by querying the same property on the children */
 354     device_class_register_property(device_class, PROPERTY_BLOCK_SIZE,
 355             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_BEFORE_START,
 356             property_get_block_size_fn,
 357             property_set_block_size_fn);
 358
 359     device_class_register_property(device_class, PROPERTY_CANONICAL_NAME,
 360             PROPERTY_ACCESS_GET_MASK,
 361             property_get_canonical_name_fn, NULL);
 362
 363     device_class_register_property(device_class, PROPERTY_CONCURRENCY,
 364             PROPERTY_ACCESS_GET_MASK,
 365             property_get_concurrency_fn, NULL);
 366
 367     device_class_register_property(device_class, PROPERTY_STREAMING,
 368             PROPERTY_ACCESS_GET_MASK,
 369             property_get_streaming_fn, NULL);
 370
 371     device_class_register_property(device_class, PROPERTY_APPENDABLE,
 372             PROPERTY_ACCESS_GET_MASK,
 373             property_get_boolean_and_fn, NULL);
 374
 375     device_class_register_property(device_class, PROPERTY_PARTIAL_DELETION,
 376             PROPERTY_ACCESS_GET_MASK,
 377             property_get_boolean_and_fn, NULL);
 378
 379     device_class_register_property(device_class, PROPERTY_FULL_DELETION,
 380             PROPERTY_ACCESS_GET_MASK,
 381             property_get_boolean_and_fn, NULL);
 382
 383     device_class_register_property(device_class, PROPERTY_LEOM,
 384             PROPERTY_ACCESS_GET_MASK,
 385             property_get_boolean_and_fn, NULL);
 386
 387     device_class_register_property(device_class, PROPERTY_MEDIUM_ACCESS_TYPE,
 388             PROPERTY_ACCESS_GET_MASK,
 389             property_get_medium_access_type_fn, NULL);
 390
 391     device_class_register_property(device_class, PROPERTY_MAX_VOLUME_USAGE,
 392             PROPERTY_ACCESS_GET_MASK | PROPERTY_ACCESS_SET_BEFORE_START,
 393             property_get_max_volume_usage_fn,
 394             property_set_max_volume_usage_fn);
 395 }
 396
 397 /* This function does something a little clever and a little
 398  * complicated. It takes an array of operations and runs the given
 399  * function on each element in the array. The trick is that it runs them
 400  * all in parallel, in different threads. This is more efficient than it
 401  * sounds because we use a GThreadPool, which means calling this function
 402  * will probably not start any new threads at all, but rather use
 403  * existing ones. The func is called with two gpointer arguments: The
 404  * first from the array, the second is the data argument.
 405  *
 406  * When it returns, all the operations have been successfully
 407  * executed. If you want results from your operations, do it yourself
 408  * through the array.
 409  */
 410
 411 #ifdef USE_INTERNAL_THREADPOOL
 412 static gpointer rait_thread_pool_func(gpointer data) {
 413     ThreadInfo *inf = data;
 414
 415     g_mutex_lock(inf->mutex);
 416     while (TRUE) {
 417         while (!inf->die && !inf->func)
 418             g_cond_wait(inf->cond, inf->mutex);
 419
 420         if (inf->die)
 421             break;
 422
 423         if (inf->func) {
 424             /* invoke the function */
 425             inf->func(inf->data, NULL);
 426             inf->func = NULL;
 427             inf->data = NULL;
 428
 429             /* indicate that we're finished; will not block */
 430             amsemaphore_down(inf->private->threads_sem);
 431         }
 432     }
 433     g_mutex_unlock(inf->mutex);
 434     return NULL;
 435 }
 436
 437 static void do_thread_pool_op(RaitDevice *self, GFunc func, GPtrArray * ops) {
 438     guint i;
 439
 440     if (PRIVATE(self)->threads_sem == NULL)
 441         PRIVATE(self)->threads_sem = amsemaphore_new_with_value(0);
 442
 443     if (PRIVATE(self)->threads == NULL)
 444         PRIVATE(self)->threads = g_array_sized_new(FALSE, TRUE,
 445                                             sizeof(ThreadInfo), ops->len);
 446
 447     g_assert(PRIVATE(self)->threads_sem->value == 0);
 448
 449     if (PRIVATE(self)->threads->len < ops->len)
 450         g_array_set_size(PRIVATE(self)->threads, ops->len);
 451
 452     /* the semaphore will hit zero when each thread has decremented it */
 453     amsemaphore_force_set(PRIVATE(self)->threads_sem, ops->len);
 454
 455     for (i = 0; i < ops->len; i++) {
 456         ThreadInfo *inf = &g_array_index(PRIVATE(self)->threads, ThreadInfo, i);
 457         if (!inf->thread) {
 458             inf->mutex = g_mutex_new();
 459             inf->cond = g_cond_new();
 460             inf->private = PRIVATE(self);
 461             inf->thread = g_thread_create(rait_thread_pool_func, inf, TRUE, NULL);
 462         }
 463
 464         /* set up the info the thread needs and trigger it to start */
 465         g_mutex_lock(inf->mutex);
 466         inf->data = g_ptr_array_index(ops, i);
 467         inf->func = func;
 468         g_cond_signal(inf->cond);
 469         g_mutex_unlock(inf->mutex);
 470     }
 471
 472     /* wait until semaphore hits zero */
 473     amsemaphore_wait_empty(PRIVATE(self)->threads_sem);
 474 }
 475
 476 #else /* USE_INTERNAL_THREADPOOL */
 477
 478 static void do_thread_pool_op(RaitDevice *self G_GNUC_UNUSED, GFunc func, GPtrArray * ops) {
 479     GThreadPool * pool;
 480     guint i;
 481
 482     pool = g_thread_pool_new(func, NULL, -1, FALSE, NULL);
 483     for (i = 0; i < ops->len; i ++) {
 484         g_thread_pool_push(pool, g_ptr_array_index(ops, i), NULL);
 485     }
 486
 487     g_thread_pool_free(pool, FALSE, TRUE);
 488 }
 489
 490 #endif /* USE_INTERNAL_THREADPOOL */
 491
 492 /* This does the above, in a serial fashion (and without using threads) */
 493 static void do_unthreaded_ops(RaitDevice *self G_GNUC_UNUSED, GFunc func, GPtrArray * ops) {
 494     guint i;
 495
 496     for (i = 0; i < ops->len; i ++) {
 497         func(g_ptr_array_index(ops, i), NULL);
 498     }
 499 }
 500
 501 /* This is the one that code below should call. It switches
 502    automatically between do_thread_pool_op and do_unthreaded_ops,
 503    depending on g_thread_supported(). */
 504 static void do_rait_child_ops(RaitDevice *self, GFunc func, GPtrArray * ops) {
 505     if (g_thread_supported()) {
 506         do_thread_pool_op(self, func, ops);
 507     } else {
 508         do_unthreaded_ops(self, func, ops);
 509     }
 510 }
 511
 512 static char *
 513 child_device_names_to_rait_name(RaitDevice * self) {
 514     GPtrArray *kids;
 515     char *braced, *result;
 516     guint i;
 517
 518     kids = g_ptr_array_sized_new(self->private->children->len);
 519     for (i = 0; i < self->private->children->len; i ++) {
 520         Device *child = g_ptr_array_index(self->private->children, i);
 521         const char *child_name = NULL;
 522         GValue val;
 523         gboolean got_prop = FALSE;
 524
 525         bzero(&val, sizeof(val));
 526
 527         if ((signed)i != self->private->failed) {
 528             if (device_property_get(child, PROPERTY_CANONICAL_NAME, &val)) {
 529                 child_name = g_value_get_string(&val);
 530                 got_prop = TRUE;
 531             }
 532         }
 533
 534         if (!got_prop)
 535             child_name = "MISSING";
 536
 537         g_ptr_array_add(kids, g_strdup(child_name));
 538
 539         if (got_prop)
 540             g_value_unset(&val);
 541     }
 542
 543     braced = collapse_braced_alternates(kids);
 544     result = g_strdup_printf("rait:%s", braced);
 545     g_free(braced);
 546
 547     return result;
 548 }
 549
 550 /* Find a workable child block size, based on the block size ranges of our
 551  * child devices.
 552  *
 553  * The algorithm is to construct the intersection of all child devices'
 554  * [min,max] block size ranges, and then pick the block size closest to 32k
 555  * that is in the resulting range.  This avoids picking ridiculously small (1
 556  * byte) or large (INT_MAX) block sizes when using devices with wide-open block
 557  * size ranges.
 558
 559  * This function returns the calculated child block size directly, and the RAIT
 560  * device's blocksize via rait_size, if not NULL.  It is resilient to errors in
 561  * a single child device, but sets the device's error status and returns 0 if
 562  * it cannot determine an agreeable block size.
 563  */
 564 static gsize
 565 calculate_block_size_from_children(RaitDevice * self, gsize *rait_size)
 566 {
 567     gsize min = 0;
 568     gsize max = SIZE_MAX;
 569     gboolean found_one = FALSE;
 570     gsize result;
 571     guint i;
 572
 573     for (i = 0; i < self->private->children->len; i ++) {
 574         gsize child_min = SIZE_MAX, child_max = 0;
 575         Device *child;
 576         GValue property_result;
 577         PropertySource source;
 578
 579         bzero(&property_result, sizeof(property_result));
 580
 581         if ((signed)i == self->private->failed)
 582             continue;
 583
 584         child = g_ptr_array_index(self->private->children, i);
 585         if (!device_property_get_ex(child, PROPERTY_BLOCK_SIZE,
 586                                  &property_result, NULL, &source)) {
 587             g_warning("Error getting BLOCK_SIZE from %s: %s",
 588                     child->device_name, device_error_or_status(child));
 589             continue;
 590         }
 591
 592         /* if the block size has been set explicitly, then we need to use that blocksize;
 593          * otherwise (even if it was DETECTED), override it. */
 594         if (source == PROPERTY_SOURCE_USER) {
 595             child_min = child_max = g_value_get_int(&property_result);
 596         } else {
 597             if (!device_property_get(child, PROPERTY_MIN_BLOCK_SIZE,
 598                                      &property_result)) {
 599                 g_warning("Error getting MIN_BLOCK_SIZE from %s: %s",
 600                         child->device_name, device_error_or_status(child));
 601                 continue;
 602             }
 603             child_min = g_value_get_uint(&property_result);
 604
 605             if (!device_property_get(child, PROPERTY_MAX_BLOCK_SIZE,
 606                                      &property_result)) {
 607                 g_warning("Error getting MAX_BLOCK_SIZE from %s: %s",
 608                         child->device_name, device_error_or_status(child));
 609                 continue;
 610             }
 611             child_max = g_value_get_uint(&property_result);
 612
 613             if (child_min == 0 || child_max == 0 || (child_min > child_max)) {
 614                 g_warning("Invalid min, max block sizes from %s", child->device_name);
 615                 continue;
 616             }
 617         }
 618
 619         found_one = TRUE;
 620         min = MAX(min, child_min);
 621         max = MIN(max, child_max);
 622     }
 623
 624     if (!found_one) {
 625         device_set_error((Device*)self,
 626             stralloc(_("Could not find any child devices' block size ranges")),
 627             DEVICE_STATUS_DEVICE_ERROR);
 628         return 0;
 629     }
 630
 631     if (min > max) {
 632         device_set_error((Device*)self,
 633             stralloc(_("No block size is acceptable to all child devices")),
 634             DEVICE_STATUS_DEVICE_ERROR);
 635         return 0;
 636     }
 637
 638     /* Now pick a number.  If 32k is in range, we use that; otherwise, we use
 639      * the nearest acceptable size. */
 640     result = CLAMP(32768, min, max);
 641
 642     if (rait_size) {
 643         guint data_children;
 644         find_simple_params(self, NULL, &data_children);
 645         *rait_size = result * data_children;
 646     }
 647
 648     return result;
 649 }
 650
 651 /* Set BLOCK_SIZE on all children */
 652 static gboolean
 653 set_block_size_on_children(RaitDevice *self, gsize child_block_size)
 654 {
 655     GValue val;
 656     guint i;
 657     PropertySource source;
 658
 659     bzero(&val, sizeof(val));
 660
 661     g_assert(child_block_size < INT_MAX);
 662     g_value_init(&val, G_TYPE_INT);
 663     g_value_set_int(&val, (gint)child_block_size);
 664
 665     for (i = 0; i < self->private->children->len; i ++) {
 666         Device *child;
 667         GValue property_result;
 668
 669         bzero(&property_result, sizeof(property_result));
 670
 671         if ((signed)i == self->private->failed)
 672             continue;
 673
 674         child = g_ptr_array_index(self->private->children, i);
 675
 676         /* first, make sure the block size is at its default, or is already
 677          * correct */
 678         if (device_property_get_ex(child, PROPERTY_BLOCK_SIZE,
 679                                  &property_result, NULL, &source)) {
 680             gsize from_child = g_value_get_int(&property_result);
 681             g_value_unset(&property_result);
 682             if (source != PROPERTY_SOURCE_DEFAULT
 683                     && from_child != child_block_size) {
 684                 device_set_error((Device *)self,
 685                     vstrallocf(_("Child device %s already has its block size set to %zd, not %zd"),
 686                                 child->device_name, from_child, child_block_size),
 687                     DEVICE_STATUS_DEVICE_ERROR);
 688                 return FALSE;
 689             }
 690         } else {
 691             /* failing to get the block size isn't necessarily fatal.. */
 692             g_warning("Error getting BLOCK_SIZE from %s: %s",
 693                     child->device_name, device_error_or_status(child));
 694         }
 695
 696         if (!device_property_set(child, PROPERTY_BLOCK_SIZE, &val)) {
 697             device_set_error((Device *)self,
 698                 vstrallocf(_("Error setting block size on %s"), child->device_name),
 699                 DEVICE_STATUS_DEVICE_ERROR);
 700             return FALSE;
 701         }
 702     }
 703
 704     return TRUE;
 705 }
 706
 707 /* The time for users to specify block sizes has ended; set this device's
 708  * block-size attributes for easy access by other RAIT functions.  Returns
 709  * FALSE on error, with the device's error status already set. */
 710 static gboolean
 711 fix_block_size(RaitDevice *self)
 712 {
 713     Device *dself = (Device *)self;
 714     gsize my_block_size, child_block_size;
 715
 716     if (dself->block_size_source == PROPERTY_SOURCE_DEFAULT) {
 717         child_block_size = calculate_block_size_from_children(self, &my_block_size);
 718         if (child_block_size == 0)
 719             return FALSE;
 720
 721         self->private->child_block_size = child_block_size;
 722         dself->block_size = my_block_size;
 723         dself->block_size_surety = PROPERTY_SURETY_GOOD;
 724         dself->block_size_source = PROPERTY_SOURCE_DETECTED;
 725     } else {
 726         guint data_children;
 727
 728         find_simple_params(self, NULL, &data_children);
 729         g_assert((dself->block_size % data_children) == 0);
 730         child_block_size = dself->block_size / data_children;
 731     }
 732
 733     /* now tell the children we mean it */
 734     if (!set_block_size_on_children(self, child_block_size))
 735         return FALSE;
 736
 737     return TRUE;
 738 }
 739
 740 /* This structure contains common fields for many operations. Not all
 741    operations use all fields, however. */
 742 typedef struct {
 743     gpointer result; /* May be a pointer; may be an integer or boolean
 744                         stored with GINT_TO_POINTER. */
 745     Device * child;  /* The device in question. Used by all
 746                         operations. */
 747     guint child_index; /* For recoverable operations (read-related
 748                           operations), this field provides the number
 749                           of this child in the self->private->children
 750                           array. */
 751 } GenericOp;
 752
 753 typedef gboolean (*BooleanExtractor)(gpointer data);
 754
 755 /* A BooleanExtractor */
 756 static gboolean extract_boolean_generic_op(gpointer data) {
 757     GenericOp * op = data;
 758     return GPOINTER_TO_INT(op->result);
 759 }
 760
 761 /* A BooleanExtractor */
 762 static gboolean extract_boolean_pointer_op(gpointer data) {
 763     GenericOp * op = data;
 764     return op->result != NULL;
 765 }
 766
 767 /* Does the equivalent of this perl command:
 768      ! (first { !extractor($_) } @_
 769    That is, calls extractor on each element of the array, and returns
 770    TRUE if and only if all calls to extractor return TRUE. This function
 771    stops as soon as an extractor returns false, so it's best if extractor
 772    functions have no side effects.
 773 */
 774 static gboolean g_ptr_array_and(GPtrArray * array,
 775                                 BooleanExtractor extractor) {
 776     guint i;
 777     if (array == NULL || array->len <= 0)
 778         return FALSE;
 779
 780     for (i = 0; i < array->len; i ++) {
 781         if (!extractor(g_ptr_array_index(array, i)))
 782             return FALSE;
 783     }
 784
 785     return TRUE;
 786 }
 787
 788 /* Takes a RaitDevice, and makes a GPtrArray of GenericOp. */
 789 static GPtrArray * make_generic_boolean_op_array(RaitDevice* self) {
 790     GPtrArray * rval;
 791     guint i;
 792
 793     rval = g_ptr_array_sized_new(self->private->children->len);
 794     for (i = 0; i < self->private->children->len; i ++) {
 795         GenericOp * op;
 796
 797         if ((signed)i == self->private->failed) {
 798             continue;
 799         }
 800
 801         op = g_new(GenericOp, 1);
 802         op->child = g_ptr_array_index(self->private->children, i);
 803         op->child_index = i;
 804         g_ptr_array_add(rval, op);
 805     }
 806
 807     return rval;
 808 }
 809
 810 /* Takes a GPtrArray of GenericOp, and a BooleanExtractor, and does
 811    all the proper handling for the result of operations that allow
 812    device isolation. Returns FALSE only if an unrecoverable error
 813    occured. */
 814 static gboolean g_ptr_array_union_robust(RaitDevice * self, GPtrArray * ops,
 815                                          BooleanExtractor extractor) {
 816     int nfailed = 0;
 817     int lastfailed = 0;
 818     guint i;
 819
 820     /* We found one or more failed elements.  See which elements failed, and
 821      * isolate them*/
 822     for (i = 0; i < ops->len; i ++) {
 823         GenericOp * op = g_ptr_array_index(ops, i);
 824         if (!extractor(op)) {
 825             self->private->failed = op->child_index;
 826             g_warning("RAIT array %s isolated device %s: %s",
 827                     DEVICE(self)->device_name,
 828                     op->child->device_name,
 829                     device_error(op->child));
 830             nfailed++;
 831             lastfailed = i;
 832         }
 833     }
 834
 835     /* no failures? great! */
 836     if (nfailed == 0)
 837         return TRUE;
 838
 839     /* a single failure in COMPLETE just puts us in DEGRADED mode */
 840     if (self->private->status == RAIT_STATUS_COMPLETE && nfailed == 1) {
 841         self->private->status = RAIT_STATUS_DEGRADED;
 842         self->private->failed = lastfailed;
 843         g_warning("RAIT array %s DEGRADED", DEVICE(self)->device_name);
 844         return TRUE;
 845     } else {
 846         self->private->status = RAIT_STATUS_FAILED;
 847         g_warning("RAIT array %s FAILED", DEVICE(self)->device_name);
 848         return FALSE;
 849     }
 850 }
 851
 852 typedef struct {
 853     RaitDevice * self;
 854     char *rait_name;
 855     char * device_name; /* IN */
 856     Device * result;    /* OUT */
 857 } OpenDeviceOp;
 858
 859 /* A GFunc. */
 860 static void device_open_do_op(gpointer data,
 861                               gpointer user_data G_GNUC_UNUSED) {
 862     OpenDeviceOp * op = data;
 863
 864     if (strcmp(op->device_name, "ERROR") == 0 ||
 865         strcmp(op->device_name, "MISSING") == 0 ||
 866         strcmp(op->device_name, "DEGRADED") == 0) {
 867         g_warning("RAIT device %s contains a missing element, attempting "
 868                   "degraded mode.\n", op->rait_name);
 869         op->result = NULL;
 870     } else {
 871         op->result = device_open(op->device_name);
 872     }
 873 }
 874
 875 /* Returns TRUE if and only if the volume label and time are equal. */
 876 static gboolean compare_volume_results(Device * a, Device * b) {
 877     return (0 == compare_possibly_null_strings(a->volume_time, b->volume_time)
 878          && 0 == compare_possibly_null_strings(a->volume_label, b->volume_label));
 879 }
 880
 881 /* Stickes new_message at the end of *old_message; frees new_message and
 882  * may change *old_message. */
 883 static void append_message(char ** old_message, char * new_message) {
 884     char * rval;
 885     if (*old_message == NULL || **old_message == '\0') {
 886         rval = new_message;
 887     } else {
 888         rval = g_strdup_printf("%s; %s", *old_message, new_message);
 889         amfree(new_message);
 890     }
 891     amfree(*old_message);
 892     *old_message = rval;
 893 }
 894
 895 static gboolean
 896 open_child_devices (Device * dself, char * device_name,
 897             char * device_node) {
 898     GPtrArray *device_names;
 899     GPtrArray * device_open_ops;
 900     guint i;
 901     gboolean failure;
 902     char *failure_errmsgs;
 903     DeviceStatusFlags failure_flags;
 904     RaitDevice * self;
 905
 906     self = RAIT_DEVICE(dself);
 907
 908     device_names = expand_braced_alternates(device_node);
 909
 910     if (device_names == NULL) {
 911         device_set_error(dself,
 912             vstrallocf(_("Invalid RAIT device name '%s'"), device_name),
 913             DEVICE_STATUS_DEVICE_ERROR);
 914         return FALSE;
 915     }
 916
 917     /* Open devices in a separate thread, in case they have to rewind etc. */
 918     device_open_ops = g_ptr_array_new();
 919
 920     for (i = 0; i < device_names->len; i++) {
 921         OpenDeviceOp *op;
 922         char *name = g_ptr_array_index(device_names, i);
 923
 924         op = g_new(OpenDeviceOp, 1);
 925         op->device_name = name;
 926         op->result = NULL;
 927         op->self = self;
 928         op->rait_name = device_name;
 929         g_ptr_array_add(device_open_ops, op);
 930     }
 931
 932     g_ptr_array_free(device_names, TRUE);
 933     do_rait_child_ops(self, device_open_do_op, device_open_ops);
 934
 935     failure = FALSE;
 936     failure_errmsgs = NULL;
 937     failure_flags = 0;
 938
 939     /* Check results of opening devices. */
 940     for (i = 0; i < device_open_ops->len; i ++) {
 941         OpenDeviceOp *op = g_ptr_array_index(device_open_ops, i);
 942
 943         if (op->result != NULL &&
 944             op->result->status == DEVICE_STATUS_SUCCESS) {
 945             g_ptr_array_add(self->private->children, op->result);
 946         } else {
 947             char * this_failure_errmsg =
 948                 g_strdup_printf("%s: %s", op->device_name,
 949                                 device_error_or_status(op->result));
 950             DeviceStatusFlags status =
 951                 op->result == NULL ?
 952                     DEVICE_STATUS_DEVICE_ERROR : op->result->status;
 953             append_message(&failure_errmsgs,
 954                            strdup(this_failure_errmsg));
 955             failure_flags |= status;
 956             if (self->private->status == RAIT_STATUS_COMPLETE) {
 957                 /* The first failure just puts us in degraded mode. */
 958                 g_warning("%s: %s",
 959                           device_name, this_failure_errmsg);
 960                 g_warning("%s: %s failed, entering degraded mode.",
 961                           device_name, op->device_name);
 962                 g_ptr_array_add(self->private->children, op->result);
 963                 self->private->status = RAIT_STATUS_DEGRADED;
 964                 self->private->failed = i;
 965             } else {
 966                 /* The second and further failures are fatal. */
 967                 failure = TRUE;
 968             }
 969         }
 970         amfree(op->device_name);
 971     }
 972
 973     g_ptr_array_free_full(device_open_ops);
 974
 975     if (failure) {
 976         self->private->status = RAIT_STATUS_FAILED;
 977         device_set_error(dself, failure_errmsgs, failure_flags);
 978         return FALSE;
 979     }
 980
 981     return TRUE;
 982 }
 983
 984 static void
 985 rait_device_open_device (Device * dself, char * device_name,
 986             char * device_type G_GNUC_UNUSED, char * device_node) {
 987
 988     if (0 != strcmp(device_node, DEFER_CHILDREN_SENTINEL)) {
 989         if (!open_child_devices(dself, device_name, device_node))
 990             return;
 991
 992         /* Chain up. */
 993         if (parent_class->open_device) {
 994             parent_class->open_device(dself, device_name, device_type, device_node);
 995         }
 996     }
 997 }
 998
 999 Device *
1000 rait_device_open_from_children (GSList *child_devices) {
1001     Device *dself;
1002     RaitDevice *self;
1003     GSList *iter;
1004     char *device_name;
1005     int nfailures;
1006     int i;
1007
1008     /* first, open a RAIT device using the DEFER_CHILDREN_SENTINEL */
1009     dself = device_open("rait:" DEFER_CHILDREN_SENTINEL);
1010     if (!IS_RAIT_DEVICE(dself)) {
1011         return dself;
1012     }
1013
1014     /* set its children */
1015     self = RAIT_DEVICE(dself);
1016     nfailures = 0;
1017     for (i=0, iter = child_devices; iter; i++, iter = iter->next) {
1018         Device *kid = iter->data;
1019
1020         /* a NULL kid is OK -- it opens the device in degraded mode */
1021         if (!kid) {
1022             nfailures++;
1023             self->private->failed = i;
1024         } else {
1025             g_assert(IS_DEVICE(kid));
1026             g_object_ref((GObject *)kid);
1027         }
1028
1029         g_ptr_array_add(self->private->children, kid);
1030     }
1031
1032     /* and set the status based on the children */
1033     switch (nfailures) {
1034         case 0:
1035             self->private->status = RAIT_STATUS_COMPLETE;
1036             break;
1037
1038         case 1:
1039             self->private->status = RAIT_STATUS_DEGRADED;
1040             break;
1041
1042         default:
1043             self->private->status = RAIT_STATUS_FAILED;
1044             device_set_error(dself,
1045                     _("more than one child device is missing"),
1046                     DEVICE_STATUS_DEVICE_ERROR);
1047             break;
1048     }
1049
1050     /* create a name from the children's names and use it to chain up
1051      * to open_device (we skipped this step in rait_device_open_device) */
1052     device_name = child_device_names_to_rait_name(self);
1053
1054     if (parent_class->open_device) {
1055         parent_class->open_device(dself,
1056             device_name, "rait",
1057             device_name+5); /* (+5 skips "rait:") */
1058     }
1059
1060     return dself;
1061 }
1062
1063 /* A GFunc. */
1064 static void read_label_do_op(gpointer data,
1065                              gpointer user_data G_GNUC_UNUSED) {
1066     GenericOp * op = data;
1067     op->result = GINT_TO_POINTER(device_read_label(op->child));
1068 }
1069
1070 static DeviceStatusFlags rait_device_read_label(Device * dself) {
1071     RaitDevice * self;
1072     GPtrArray * ops;
1073     DeviceStatusFlags failed_result = 0;
1074     char *failed_errmsg = NULL;
1075     unsigned int i;
1076     Device * first_success = NULL;
1077
1078     self = RAIT_DEVICE(dself);
1079
1080     amfree(dself->volume_time);
1081     amfree(dself->volume_label);
1082     dumpfile_free(dself->volume_header);
1083     dself->volume_header = NULL;
1084
1085     if (rait_device_in_error(self))
1086         return dself->status | DEVICE_STATUS_DEVICE_ERROR;
1087
1088     /* nail down our block size, if we haven't already */
1089     if (!fix_block_size(self))
1090         return FALSE;
1091
1092     ops = make_generic_boolean_op_array(self);
1093
1094     do_rait_child_ops(self, read_label_do_op, ops);
1095
1096     for (i = 0; i < ops->len; i ++) {
1097         GenericOp * op = g_ptr_array_index(ops, i);
1098         DeviceStatusFlags result = GPOINTER_TO_INT(op->result);
1099         if (op->result == DEVICE_STATUS_SUCCESS) {
1100             if (first_success == NULL) {
1101                 /* This is the first successful device. */
1102                 first_success = op->child;
1103             } else if (!compare_volume_results(first_success, op->child)) {
1104                 /* Doesn't match. :-( */
1105                 failed_errmsg = vstrallocf("Inconsistent volume labels/datestamps: "
1106                         "Got %s/%s on %s against %s/%s on %s.",
1107                         first_success->volume_label,
1108                         first_success->volume_time,
1109                         first_success->device_name,
1110                         op->child->volume_label,
1111                         op->child->volume_time,
1112                         op->child->device_name);
1113                 g_warning("%s", failed_errmsg);
1114                 failed_result |= DEVICE_STATUS_VOLUME_ERROR;
1115             }
1116         } else {
1117             failed_result |= result;
1118         }
1119     }
1120
1121     if (failed_result != DEVICE_STATUS_SUCCESS) {
1122         /* We had multiple failures or an inconsistency. */
1123         device_set_error(dself, failed_errmsg, failed_result);
1124     } else {
1125         /* Everything peachy. */
1126         amfree(failed_errmsg);
1127
1128         g_assert(first_success != NULL);
1129         if (first_success->volume_label != NULL) {
1130             dself->volume_label = g_strdup(first_success->volume_label);
1131         }
1132         if (first_success->volume_time != NULL) {
1133             dself->volume_time = g_strdup(first_success->volume_time);
1134         }
1135         if (first_success->volume_header != NULL) {
1136             dself->volume_header = dumpfile_copy(first_success->volume_header);
1137         }
1138         dself->header_block_size = first_success->header_block_size;
1139     }
1140
1141     g_ptr_array_free_full(ops);
1142
1143     return dself->status;
1144 }
1145
1146 typedef struct {
1147     GenericOp base;
1148     DeviceAccessMode mode; /* IN */
1149     char * label;          /* IN */
1150     char * timestamp;      /* IN */
1151 } StartOp;
1152
1153 /* A GFunc. */
1154 static void start_do_op(gpointer data, gpointer user_data G_GNUC_UNUSED) {
1155     DeviceClass *klass;
1156     StartOp * param = data;
1157
1158     klass = DEVICE_GET_CLASS(param->base.child);
1159     if (klass->start) {
1160         param->base.result =
1161             GINT_TO_POINTER((klass->start)(param->base.child,
1162                                             param->mode, param->label,
1163                                             param->timestamp));
1164     } else {
1165         param->base.result = FALSE;
1166     }
1167 }
1168
1169 static gboolean
1170 rait_device_configure(Device * dself, gboolean use_global_config)
1171 {
1172     RaitDevice *self = RAIT_DEVICE(dself);
1173     guint i;
1174
1175     for (i = 0; i < self->private->children->len; i ++) {
1176         Device *child;
1177
1178         if ((signed)i == self->private->failed)
1179             continue;
1180
1181         child = g_ptr_array_index(self->private->children, i);
1182         /* unconditionally configure the child without the global
1183          * configuration */
1184         if (!device_configure(child, FALSE))
1185             return FALSE;
1186     }
1187
1188     if (parent_class->configure) {
1189         return parent_class->configure(dself, use_global_config);
1190     }
1191
1192     return TRUE;
1193 }
1194
1195 static gboolean
1196 rait_device_start (Device * dself, DeviceAccessMode mode, char * label,
1197                    char * timestamp) {
1198     GPtrArray * ops;
1199     guint i;
1200     gboolean success;
1201     RaitDevice * self;
1202     DeviceStatusFlags total_status;
1203     char *failure_errmsgs = NULL;
1204     char * label_from_device = NULL;
1205
1206     self = RAIT_DEVICE(dself);
1207
1208     if (rait_device_in_error(self)) return FALSE;
1209
1210     /* No starting in degraded mode. */
1211     if (self->private->status != RAIT_STATUS_COMPLETE &&
1212         (mode == ACCESS_WRITE || mode == ACCESS_APPEND)) {
1213         device_set_error(dself,
1214                          g_strdup_printf(_("RAIT device %s is read-only "
1215                                            "because it is in degraded mode.\n"),
1216                                          dself->device_name),
1217                          DEVICE_STATUS_DEVICE_ERROR);
1218         return FALSE;
1219     }
1220
1221     /* nail down our block size, if we haven't already */
1222     if (!fix_block_size(self))
1223         return FALSE;
1224
1225     dself->access_mode = mode;
1226     g_mutex_lock(dself->device_mutex);
1227     dself->in_file = FALSE;
1228     g_mutex_unlock(dself->device_mutex);
1229     amfree(dself->volume_label);
1230     amfree(dself->volume_time);
1231     dumpfile_free(dself->volume_header);
1232     dself->volume_header = NULL;
1233
1234     ops = g_ptr_array_sized_new(self->private->children->len);
1235     for (i = 0; i < self->private->children->len; i ++) {
1236         StartOp * op;
1237
1238         if ((signed)i == self->private->failed) {
1239             continue;
1240         }
1241
1242         op = g_new(StartOp, 1);
1243         op->base.child = g_ptr_array_index(self->private->children, i);
1244         op->mode = mode;
1245         op->label = g_strdup(label);
1246         op->timestamp = g_strdup(timestamp);
1247         g_ptr_array_add(ops, op);
1248     }
1249
1250     do_rait_child_ops(self, start_do_op, ops);
1251
1252     success = g_ptr_array_and(ops, extract_boolean_generic_op);
1253
1254     /* Check results of starting devices; this is mostly about the
1255      * VOLUME_UNLABELED flag. */
1256     total_status = 0;
1257     for (i = 0; i < ops->len; i ++) {
1258         StartOp * op = g_ptr_array_index(ops, i);
1259         Device *child = op->base.child;
1260
1261         total_status |= child->status;
1262         if (child->status != DEVICE_STATUS_SUCCESS) {
1263             /* record the error message and move on. */
1264             append_message(&failure_errmsgs,
1265                            g_strdup_printf("%s: %s",
1266                                            child->device_name,
1267                                            device_error_or_status(child)));
1268         } else {
1269             if (child->volume_label != NULL && child->volume_time != NULL) {
1270                 if (label_from_device) {
1271                     if (strcmp(child->volume_label, dself->volume_label) != 0 ||
1272                         strcmp(child->volume_time, dself->volume_time) != 0) {
1273                         /* Mismatch! (Two devices provided different labels) */
1274                         char * this_message =
1275                             g_strdup_printf("%s: Label (%s/%s) is different "
1276                                             "from label (%s/%s) found at "
1277                                             "device %s",
1278                                             child->device_name,
1279                                             child->volume_label,
1280                                             child->volume_time,
1281                                             dself->volume_label,
1282                                             dself->volume_time,
1283                                             label_from_device);
1284                         append_message(&failure_errmsgs, this_message);
1285                         total_status |= DEVICE_STATUS_DEVICE_ERROR;
1286                         g_warning("RAIT device children have different labels or timestamps");
1287                     }
1288                 } else {
1289                     /* First device with a volume. */
1290                     dself->volume_label = g_strdup(child->volume_label);
1291                     dself->volume_time = g_strdup(child->volume_time);
1292                     dself->volume_header = dumpfile_copy(child->volume_header);
1293                     label_from_device = g_strdup(child->device_name);
1294                 }
1295             } else {
1296                 /* Device problem, it says it succeeded but sets no label? */
1297                 char * this_message =
1298                     g_strdup_printf("%s: Says label read, but no volume "
1299                                      "label found.", child->device_name);
1300                 g_warning("RAIT device child has NULL volume or label");
1301                 append_message(&failure_errmsgs, this_message);
1302                 total_status |= DEVICE_STATUS_DEVICE_ERROR;
1303             }
1304         }
1305     }
1306
1307     if (total_status == DEVICE_STATUS_SUCCESS) {
1308         StartOp * op = g_ptr_array_index(ops, 0);
1309         Device *child = op->base.child;
1310         dself->header_block_size = child->header_block_size;
1311     }
1312
1313     amfree(label_from_device);
1314     g_ptr_array_free_full(ops);
1315
1316     dself->status = total_status;
1317
1318     if (total_status != DEVICE_STATUS_SUCCESS || !success) {
1319         device_set_error(dself, failure_errmsgs, total_status);
1320         return FALSE;
1321     }
1322     amfree(failure_errmsgs);
1323     return TRUE;
1324 }
1325
1326 typedef struct {
1327     GenericOp base;
1328     dumpfile_t * info; /* IN */
1329     int fileno;
1330 } StartFileOp;
1331
1332 /* a GFunc */
1333 static void start_file_do_op(gpointer data, gpointer user_data G_GNUC_UNUSED) {
1334     StartFileOp * op = data;
1335     op->base.result = GINT_TO_POINTER(device_start_file(op->base.child,
1336                                                         op->info));
1337     op->fileno = op->base.child->file;
1338     if (op->fileno < 1) {
1339         op->base.result = FALSE;
1340     }
1341 }
1342
1343 static gboolean
1344 rait_device_start_file (Device * dself, dumpfile_t * info) {
1345     GPtrArray * ops;
1346     guint i;
1347     gboolean success;
1348     RaitDevice * self;
1349     int actual_file = -1;
1350
1351     self = RAIT_DEVICE(dself);
1352
1353     if (rait_device_in_error(self)) return FALSE;
1354     if (self->private->status != RAIT_STATUS_COMPLETE) return FALSE;
1355
1356     ops = g_ptr_array_sized_new(self->private->children->len);
1357     for (i = 0; i < self->private->children->len; i ++) {
1358         StartFileOp * op;
1359         op = g_new(StartFileOp, 1);
1360         op->base.child = g_ptr_array_index(self->private->children, i);
1361         /* each child gets its own copy of the header, to munge as it
1362          * likes (setting blocksize, at least) */
1363         op->info = dumpfile_copy(info);
1364         g_ptr_array_add(ops, op);
1365     }
1366
1367     do_rait_child_ops(self, start_file_do_op, ops);
1368
1369     success = g_ptr_array_and(ops, extract_boolean_generic_op);
1370
1371     for (i = 0; i < self->private->children->len && success; i ++) {
1372         StartFileOp * op = g_ptr_array_index(ops, i);
1373         if (!op->base.result)
1374             continue;
1375         g_assert(op->fileno >= 1);
1376         if (actual_file < 1) {
1377             actual_file = op->fileno;
1378         }
1379         if (actual_file != op->fileno) {
1380             /* File number mismatch! Aah, my hair is on fire! */
1381             device_set_error(dself,
1382                              g_strdup_printf("File number mismatch in "
1383                                              "rait_device_start_file(): "
1384                                              "Child %s reported file number "
1385                                              "%d, another child reported "
1386                                              "file number %d.",
1387                                              op->base.child->device_name,
1388                                              op->fileno, actual_file),
1389                              DEVICE_STATUS_DEVICE_ERROR);
1390             success = FALSE;
1391             op->base.result = FALSE;
1392         }
1393     }
1394
1395     for (i = 0; i < ops->len && success; i ++) {
1396         StartFileOp * op = g_ptr_array_index(ops, i);
1397         if (op->info) dumpfile_free(op->info);
1398     }
1399     g_ptr_array_free_full(ops);
1400
1401     if (!success) {
1402         if (!device_in_error(dself)) {
1403             device_set_error(dself, stralloc("One or more devices "
1404                                              "failed to start_file"),
1405                              DEVICE_STATUS_DEVICE_ERROR);
1406         }
1407         return FALSE;
1408     }
1409
1410     g_assert(actual_file >= 1);
1411     dself->file = actual_file;
1412     g_mutex_lock(dself->device_mutex);
1413     dself->in_file = TRUE;
1414     dself->bytes_written = 0;
1415     g_mutex_unlock(dself->device_mutex);
1416
1417     return TRUE;
1418 }
1419
1420 static void find_simple_params(RaitDevice * self,
1421                                guint * num_children,
1422                                guint * data_children) {
1423     int num, data;
1424
1425     num = self->private->children->len;
1426     if (num > 1)
1427         data = num - 1;
1428     else
1429         data = num;
1430     if (num_children != NULL)
1431         *num_children = num;
1432     if (data_children != NULL)
1433         *data_children = data;
1434 }
1435
1436 typedef struct {
1437     GenericOp base;
1438     guint size;           /* IN */
1439     gpointer data;        /* IN */
1440     gboolean data_needs_free; /* bookkeeping */
1441 } WriteBlockOp;
1442
1443 /* a GFunc. */
1444 static void write_block_do_op(gpointer data,
1445                               gpointer user_data G_GNUC_UNUSED) {
1446     WriteBlockOp * op = data;
1447
1448     op->base.result =
1449         GINT_TO_POINTER(device_write_block(op->base.child, op->size, op->data));
1450 }
1451
1452 /* Parity block generation. Performance of this function can be improved
1453    considerably by using larger-sized integers or
1454    assembly-coded vector instructions. Parameters are:
1455    % data       - All data chunks in series (chunk_size * num_chunks bytes)
1456    % parity     - Allocated space for parity block (chunk_size bytes)
1457  */
1458 static void make_parity_block(char * data, char * parity,
1459                               guint chunk_size, guint num_chunks) {
1460     guint i;
1461     bzero(parity, chunk_size);
1462     for (i = 0; i < num_chunks - 1; i ++) {
1463         guint j;
1464         for (j = 0; j < chunk_size; j ++) {
1465             parity[j] ^= data[chunk_size*i + j];
1466         }
1467     }
1468 }
1469
1470 /* Does the same thing as make_parity_block, but instead of using a
1471    single memory chunk holding all chunks, it takes a GPtrArray of
1472    chunks. */
1473 static void make_parity_block_extents(GPtrArray * data, char * parity,
1474                                       guint chunk_size) {
1475     guint i;
1476     bzero(parity, chunk_size);
1477     for (i = 0; i < data->len; i ++) {
1478         guint j;
1479         char * data_chunk;
1480         data_chunk = g_ptr_array_index(data, i);
1481         for (j = 0; j < chunk_size; j ++) {
1482             parity[j] ^= data_chunk[j];
1483         }
1484     }
1485 }
1486
1487 /* Does the parity creation algorithm. Allocates and returns a single
1488    device block from a larger RAIT block. chunks and chunk are 1-indexed. */
1489 static char * extract_data_block(char * data, guint size,
1490                                  guint chunks, guint chunk) {
1491     char * rval;
1492     guint chunk_size;
1493
1494     g_assert(chunks > 0 && chunk > 0 && chunk <= chunks);
1495     g_assert(data != NULL);
1496     g_assert(size > 0 && size % (chunks - 1) == 0);
1497
1498     chunk_size = size / (chunks - 1);
1499     rval = g_malloc(chunk_size);
1500     if (chunks != chunk) {
1501         /* data block. */
1502         memcpy(rval, data + chunk_size * (chunk - 1), chunk_size);
1503     } else {
1504         make_parity_block(data, rval, chunk_size, chunks);
1505     }
1506
1507     return rval;
1508 }
1509
1510 static gboolean
1511 rait_device_write_block (Device * dself, guint size, gpointer data) {
1512     GPtrArray * ops;
1513     guint i;
1514     gboolean success;
1515     guint data_children, num_children;
1516     gsize blocksize = dself->block_size;
1517     RaitDevice * self;
1518     gboolean last_block = (size < blocksize);
1519
1520     self = RAIT_DEVICE(dself);
1521
1522     if (rait_device_in_error(self)) return FALSE;
1523     if (self->private->status != RAIT_STATUS_COMPLETE) return FALSE;
1524
1525     find_simple_params(RAIT_DEVICE(self), &num_children, &data_children);
1526     num_children = self->private->children->len;
1527     if (num_children != 1)
1528         data_children = num_children - 1;
1529     else
1530         data_children = num_children;
1531
1532     g_assert(size % data_children == 0 || last_block);
1533
1534     /* zero out to the end of a short block -- tape devices only write
1535      * whole blocks. */
1536     if (last_block) {
1537         char *new_data;
1538
1539         new_data = g_malloc(blocksize);
1540         memcpy(new_data, data, size);
1541         bzero(new_data + size, blocksize - size);
1542
1543         data = new_data;
1544         size = blocksize;
1545     }
1546
1547     ops = g_ptr_array_sized_new(num_children);
1548     for (i = 0; i < self->private->children->len; i ++) {
1549         WriteBlockOp * op;
1550         op = g_malloc(sizeof(*op));
1551         op->base.child = g_ptr_array_index(self->private->children, i);
1552         op->size = size / data_children;
1553         if (num_children <= 2) {
1554             op->data = data;
1555             op->data_needs_free = FALSE;
1556         } else {
1557             op->data_needs_free = TRUE;
1558             op->data = extract_data_block(data, size, num_children, i + 1);
1559         }
1560         g_ptr_array_add(ops, op);
1561     }
1562
1563     do_rait_child_ops(self, write_block_do_op, ops);
1564
1565     success = g_ptr_array_and(ops, extract_boolean_generic_op);
1566
1567     for (i = 0; i < self->private->children->len; i ++) {
1568         WriteBlockOp * op = g_ptr_array_index(ops, i);
1569         if (op->data_needs_free)
1570             free(op->data);
1571     }
1572
1573     if (last_block) {
1574         amfree(data);
1575     }
1576
1577     g_ptr_array_free_full(ops);
1578
1579     if (!success) {
1580         /* TODO be more specific here */
1581         /* TODO: handle EOM here -- if one or more (or two or more??)
1582          * children have is_eom set, then reflect that in our error
1583          * status. What's more fun is when one device fails and must be isolated at
1584          * the same time another hits EOF. */
1585         device_set_error(dself,
1586             stralloc("One or more devices failed to write_block"),
1587             DEVICE_STATUS_DEVICE_ERROR);
1588         /* this is EOM or an error, so call it EOM */
1589         dself->is_eom = TRUE;
1590         return FALSE;
1591     } else {
1592         dself->block ++;
1593         g_mutex_lock(dself->device_mutex);
1594         dself->bytes_written += size;
1595         g_mutex_unlock(dself->device_mutex);
1596
1597         return TRUE;
1598     }
1599 }
1600
1601 /* A GFunc */
1602 static void finish_file_do_op(gpointer data,
1603                               gpointer user_data G_GNUC_UNUSED) {
1604     GenericOp * op = data;
1605     if (op->child) {
1606         op->result = GINT_TO_POINTER(device_finish_file(op->child));
1607     } else {
1608         op->result = FALSE;
1609     }
1610 }
1611
1612 static gboolean
1613 rait_device_finish_file (Device * dself) {
1614     GPtrArray * ops;
1615     gboolean success;
1616     RaitDevice * self = RAIT_DEVICE(dself);
1617
1618     g_assert(self != NULL);
1619     if (rait_device_in_error(dself)) return FALSE;
1620     if (self->private->status != RAIT_STATUS_COMPLETE) return FALSE;
1621
1622     ops = make_generic_boolean_op_array(self);
1623
1624     do_rait_child_ops(self, finish_file_do_op, ops);
1625
1626     success = g_ptr_array_and(ops, extract_boolean_generic_op);
1627
1628     g_ptr_array_free_full(ops);
1629
1630     if (!success) {
1631         /* TODO: be more specific here */
1632         device_set_error(dself,
1633                          g_strdup("One or more devices failed to finish_file"),
1634             DEVICE_STATUS_DEVICE_ERROR);
1635         return FALSE;
1636     }
1637
1638     g_mutex_lock(dself->device_mutex);
1639     dself->in_file = FALSE;
1640     g_mutex_unlock(dself->device_mutex);
1641     return TRUE;
1642 }
1643
1644 typedef struct {
1645     GenericOp base;
1646     guint requested_file;                /* IN */
1647     guint actual_file;                   /* OUT */
1648 } SeekFileOp;
1649
1650 /* a GFunc. */
1651 static void seek_file_do_op(gpointer data, gpointer user_data G_GNUC_UNUSED) {
1652     SeekFileOp * op = data;
1653     op->base.result = device_seek_file(op->base.child, op->requested_file);
1654     op->actual_file = op->base.child->file;
1655 }
1656
1657 static dumpfile_t *
1658 rait_device_seek_file (Device * dself, guint file) {
1659     GPtrArray * ops;
1660     guint i;
1661     gboolean success;
1662     dumpfile_t * rval;
1663     RaitDevice * self = RAIT_DEVICE(dself);
1664     guint actual_file = 0;
1665     gboolean in_file = FALSE;
1666
1667     if (rait_device_in_error(self)) return NULL;
1668
1669     dself->is_eof = FALSE;
1670     dself->block = 0;
1671     g_mutex_lock(dself->device_mutex);
1672     dself->in_file = FALSE;
1673     dself->bytes_read = 0;
1674     g_mutex_unlock(dself->device_mutex);
1675
1676     ops = g_ptr_array_sized_new(self->private->children->len);
1677     for (i = 0; i < self->private->children->len; i ++) {
1678         SeekFileOp * op;
1679         if ((int)i == self->private->failed)
1680             continue; /* This device is broken. */
1681         op = g_new(SeekFileOp, 1);
1682         op->base.child = g_ptr_array_index(self->private->children, i);
1683         op->base.child_index = i;
1684         op->requested_file = file;
1685         g_ptr_array_add(ops, op);
1686     }
1687
1688     do_rait_child_ops(self, seek_file_do_op, ops);
1689
1690     /* This checks for NULL values, but we still have to check for
1691        consistant headers. */
1692     success = g_ptr_array_union_robust(RAIT_DEVICE(self),
1693                                        ops, extract_boolean_pointer_op);
1694
1695     rval = NULL;
1696     for (i = 0; i < ops->len; i ++) {
1697         SeekFileOp * this_op;
1698         dumpfile_t * this_result;
1699         guint this_actual_file;
1700         gboolean this_in_file;
1701
1702         this_op = (SeekFileOp*)g_ptr_array_index(ops, i);
1703
1704         if ((signed)this_op->base.child_index == self->private->failed)
1705             continue;
1706
1707         this_result = this_op->base.result;
1708         this_actual_file = this_op->actual_file;
1709         this_in_file = this_op->base.child->in_file;
1710
1711         if (rval == NULL) {
1712             rval = this_result;
1713             actual_file = this_actual_file;
1714             in_file = this_in_file;
1715         } else {
1716             if (headers_are_equal(rval, this_result) &&
1717                 actual_file == this_actual_file &&
1718                 in_file == this_in_file) {
1719                 /* Do nothing. */
1720             } else {
1721                 success = FALSE;
1722             }
1723             free(this_result);
1724         }
1725     }
1726
1727     g_ptr_array_free_full(ops);
1728
1729     if (!success) {
1730         amfree(rval);
1731         /* TODO: be more specific here */
1732         device_set_error(dself,
1733                          g_strdup("One or more devices failed to seek_file"),
1734             DEVICE_STATUS_DEVICE_ERROR);
1735         return NULL;
1736     }
1737
1738     /* update our state */
1739     g_mutex_lock(dself->device_mutex);
1740     dself->in_file = in_file;
1741     g_mutex_unlock(dself->device_mutex);
1742     dself->file = actual_file;
1743
1744     return rval;
1745 }
1746
1747 typedef struct {
1748     GenericOp base;
1749     guint64 block; /* IN */
1750 } SeekBlockOp;
1751
1752 /* a GFunc. */
1753 static void seek_block_do_op(gpointer data, gpointer user_data G_GNUC_UNUSED) {
1754     SeekBlockOp * op = data;
1755     op->base.result =
1756         GINT_TO_POINTER(device_seek_block(op->base.child, op->block));
1757 }
1758
1759 static gboolean
1760 rait_device_seek_block (Device * dself, guint64 block) {
1761     GPtrArray * ops;
1762     guint i;
1763     gboolean success;
1764
1765     RaitDevice * self = RAIT_DEVICE(dself);
1766
1767     if (rait_device_in_error(self)) return FALSE;
1768
1769     ops = g_ptr_array_sized_new(self->private->children->len);
1770     for (i = 0; i < self->private->children->len; i ++) {
1771         SeekBlockOp * op;
1772         if ((int)i == self->private->failed)
1773             continue; /* This device is broken. */
1774         op = g_new(SeekBlockOp, 1);
1775         op->base.child = g_ptr_array_index(self->private->children, i);
1776         op->base.child_index = i;
1777         op->block = block;
1778         g_ptr_array_add(ops, op);
1779     }
1780
1781     do_rait_child_ops(self, seek_block_do_op, ops);
1782
1783     success = g_ptr_array_union_robust(RAIT_DEVICE(self),
1784                                        ops, extract_boolean_generic_op);
1785
1786     g_ptr_array_free_full(ops);
1787
1788     if (!success) {
1789         /* TODO: be more specific here */
1790         device_set_error(dself,
1791             stralloc("One or more devices failed to seek_block"),
1792             DEVICE_STATUS_DEVICE_ERROR);
1793         return FALSE;
1794     }
1795
1796     dself->block = block;
1797     return TRUE;
1798 }
1799
1800 typedef struct {
1801     GenericOp base;
1802     gpointer buffer; /* IN */
1803     int read_size;      /* IN/OUT -- note not a pointer */
1804     int desired_read_size; /* bookkeeping */
1805 } ReadBlockOp;
1806
1807 /* a GFunc. */
1808 static void read_block_do_op(gpointer data,
1809                              gpointer user_data G_GNUC_UNUSED) {
1810     ReadBlockOp * op = data;
1811     op->base.result =
1812         GINT_TO_POINTER(device_read_block(op->base.child, op->buffer,
1813                                           &(op->read_size)));
1814     if (op->read_size > op->desired_read_size) {
1815         g_warning("child device %s tried to return an oversized block, which the RAIT device does not support",
1816                   op->base.child->device_name);
1817     }
1818 }
1819
1820 /* A BooleanExtractor. This one checks for a successful read. */
1821 static gboolean extract_boolean_read_block_op_data(gpointer data) {
1822     ReadBlockOp * op = data;
1823     return GPOINTER_TO_INT(op->base.result) == op->desired_read_size;
1824 }
1825
1826 /* A BooleanExtractor. This one checks for EOF. */
1827 static gboolean extract_boolean_read_block_op_eof(gpointer data) {
1828     ReadBlockOp * op = data;
1829     return op->base.child->is_eof;
1830 }
1831
1832 /* Counts the number of elements in an array matching a given proposition. */
1833 static int g_ptr_array_count(GPtrArray * array, BooleanExtractor filter) {
1834     int rval;
1835     unsigned int i;
1836     rval = 0;
1837     for (i = 0; i < array->len ; i++) {
1838         if (filter(g_ptr_array_index(array, i)))
1839             rval ++;
1840     }
1841     return rval;
1842 }
1843
1844 static gboolean raid_block_reconstruction(RaitDevice * self, GPtrArray * ops,
1845                                       gpointer buf, size_t bufsize) {
1846     guint num_children, data_children;
1847     gsize blocksize;
1848     gsize child_blocksize;
1849     guint i;
1850     int parity_child;
1851     gpointer parity_block = NULL;
1852     gboolean success;
1853
1854     success = TRUE;
1855
1856     blocksize = DEVICE(self)->block_size;
1857     find_simple_params(self, &num_children, &data_children);
1858
1859     if (num_children > 1)
1860         parity_child = num_children - 1;
1861     else
1862         parity_child = -1;
1863
1864     child_blocksize = blocksize / data_children;
1865
1866     for (i = 0; i < ops->len; i ++) {
1867         ReadBlockOp * op = g_ptr_array_index(ops, i);
1868         if (!extract_boolean_read_block_op_data(op))
1869             continue;
1870         if ((int)(op->base.child_index) == parity_child) {
1871             parity_block = op->buffer;
1872         } else {
1873             g_assert(child_blocksize * (op->base.child_index+1) <= bufsize);
1874             memcpy((char *)buf + child_blocksize * op->base.child_index, op->buffer,
1875                    child_blocksize);
1876         }
1877     }
1878     if (self->private->status == RAIT_STATUS_COMPLETE) {
1879         g_assert(parity_block != NULL); /* should have found parity_child */
1880
1881         if (num_children >= 2) {
1882             /* Verify the parity block. This code is inefficient but
1883                does the job for the 2-device case, too. */
1884             gpointer constructed_parity;
1885             GPtrArray * data_extents;
1886
1887             constructed_parity = g_malloc(child_blocksize);
1888             data_extents = g_ptr_array_sized_new(data_children);
1889             for (i = 0; i < data_children; i ++) {
1890                 ReadBlockOp * op = g_ptr_array_index(ops, i);
1891                 g_assert(extract_boolean_read_block_op_data(op));
1892                 if ((int)op->base.child_index == parity_child)
1893                     continue;
1894                 g_ptr_array_add(data_extents, op->buffer);
1895             }
1896             make_parity_block_extents(data_extents, constructed_parity,
1897                                       child_blocksize);
1898
1899             if (0 != memcmp(parity_block, constructed_parity,
1900                             child_blocksize)) {
1901                 device_set_error(DEVICE(self),
1902                     stralloc(_("RAIT is inconsistent: Parity block did not match data blocks.")),
1903                     DEVICE_STATUS_DEVICE_ERROR);
1904                 /* TODO: can't we just isolate the device in this case? */
1905                 success = FALSE;
1906             }
1907             g_ptr_array_free(data_extents, TRUE);
1908             amfree(constructed_parity);
1909         } else { /* do nothing. */ }
1910     } else if (self->private->status == RAIT_STATUS_DEGRADED) {
1911         g_assert(self->private->failed >= 0 && self->private->failed < (int)num_children);
1912         /* We are in degraded mode. What's missing? */
1913         if (self->private->failed == parity_child) {
1914             /* do nothing. */
1915         } else if (num_children >= 2) {
1916             /* Reconstruct failed block from parity block. */
1917             GPtrArray * data_extents = g_ptr_array_new();
1918
1919             for (i = 0; i < data_children; i ++) {
1920                 ReadBlockOp * op = g_ptr_array_index(ops, i);
1921                 if (!extract_boolean_read_block_op_data(op))
1922                     continue;
1923                 g_ptr_array_add(data_extents, op->buffer);
1924             }
1925
1926             /* Conveniently, the reconstruction is the same procedure
1927                as the parity generation. This even works if there is
1928                only one remaining device! */
1929             make_parity_block_extents(data_extents,
1930                                       (char *)buf + (child_blocksize *
1931                                              self->private->failed),
1932                                       child_blocksize);
1933
1934             /* The array members belong to our ops argument. */
1935             g_ptr_array_free(data_extents, TRUE);
1936         } else {
1937             g_assert_not_reached();
1938         }
1939     } else {
1940         /* device is already in FAILED state -- we shouldn't even be here */
1941         success = FALSE;
1942     }
1943     return success;
1944 }
1945
1946 static int
1947 rait_device_read_block (Device * dself, gpointer buf, int * size) {
1948     GPtrArray * ops;
1949     guint i;
1950     gboolean success;
1951     guint num_children, data_children;
1952     gsize blocksize = dself->block_size;
1953     gsize child_blocksize;
1954
1955     RaitDevice * self = RAIT_DEVICE(dself);
1956
1957     if (rait_device_in_error(self)) return -1;
1958
1959     find_simple_params(self, &num_children, &data_children);
1960
1961     /* tell caller they haven't given us a big enough buffer */
1962     if (blocksize > (gsize)*size) {
1963         g_assert(blocksize < INT_MAX);
1964         *size = (int)blocksize;
1965         return 0;
1966     }
1967
1968     g_assert(blocksize % data_children == 0); /* see find_block_size */
1969     child_blocksize = blocksize / data_children;
1970
1971     ops = g_ptr_array_sized_new(num_children);
1972     for (i = 0; i < num_children; i ++) {
1973         ReadBlockOp * op;
1974         if ((int)i == self->private->failed)
1975             continue; /* This device is broken. */
1976         op = g_new(ReadBlockOp, 1);
1977         op->base.child = g_ptr_array_index(self->private->children, i);
1978         op->base.child_index = i;
1979         op->buffer = g_malloc(child_blocksize);
1980         op->desired_read_size = op->read_size = child_blocksize;
1981         g_ptr_array_add(ops, op);
1982     }
1983
1984     do_rait_child_ops(self, read_block_do_op, ops);
1985
1986     if (g_ptr_array_count(ops, extract_boolean_read_block_op_data)) {
1987         if (!g_ptr_array_union_robust(RAIT_DEVICE(self),
1988                                      ops,
1989                                      extract_boolean_read_block_op_data)) {
1990             /* TODO: be more specific */
1991             device_set_error(dself,
1992                 stralloc(_("Error occurred combining blocks from child devices")),
1993                 DEVICE_STATUS_DEVICE_ERROR);
1994             success = FALSE;
1995         } else {
1996             /* raid_block_reconstruction sets the error status if necessary */
1997             success = raid_block_reconstruction(RAIT_DEVICE(self),
1998                                                 ops, buf, (size_t)*size);
1999         }
2000     } else {
2001         success = FALSE;
2002         if (g_ptr_array_union_robust(RAIT_DEVICE(self),
2003                                      ops,
2004                                      extract_boolean_read_block_op_eof)) {
2005             device_set_error(dself,
2006                 stralloc(_("EOF")),
2007                 DEVICE_STATUS_SUCCESS);
2008             dself->is_eof = TRUE;
2009             g_mutex_lock(dself->device_mutex);
2010             dself->in_file = FALSE;
2011             g_mutex_unlock(dself->device_mutex);
2012         } else {
2013             device_set_error(dself,
2014                 stralloc(_("All child devices failed to read, but not all are at eof")),
2015                 DEVICE_STATUS_DEVICE_ERROR);
2016         }
2017     }
2018
2019     for (i = 0; i < ops->len; i ++) {
2020         ReadBlockOp * op = g_ptr_array_index(ops, i);
2021         amfree(op->buffer);
2022     }
2023     g_ptr_array_free_full(ops);
2024
2025     if (success) {
2026         dself->block++;
2027         *size = blocksize;
2028         g_mutex_lock(dself->device_mutex);
2029         dself->bytes_read += blocksize;
2030         g_mutex_unlock(dself->device_mutex);
2031         return blocksize;
2032     } else {
2033         return -1;
2034     }
2035 }
2036
2037 /* property utility functions */
2038
2039 typedef struct {
2040     GenericOp base;
2041     DevicePropertyId id;   /* IN */
2042     GValue value;          /* IN/OUT */
2043     PropertySurety surety; /* IN (for set) */
2044     PropertySource source; /* IN (for set) */
2045 } PropertyOp;
2046
2047 /* Creates a GPtrArray of PropertyOf for a get or set operation. */
2048 static GPtrArray * make_property_op_array(RaitDevice * self,
2049                                           DevicePropertyId id,
2050                                           GValue * value,
2051                                           PropertySurety surety,
2052                                           PropertySource source) {
2053     guint i;
2054     GPtrArray * ops;
2055     ops = g_ptr_array_sized_new(self->private->children->len);
2056     for (i = 0; i < self->private->children->len; i ++) {
2057         PropertyOp * op;
2058
2059         if ((signed)i == self->private->failed) {
2060             continue;
2061         }
2062
2063         op = g_new(PropertyOp, 1);
2064         op->base.child = g_ptr_array_index(self->private->children, i);
2065         op->id = id;
2066         bzero(&(op->value), sizeof(op->value));
2067         if (value != NULL) {
2068             g_value_unset_copy(value, &(op->value));
2069         }
2070         op->surety = surety;
2071         op->source = source;
2072         g_ptr_array_add(ops, op);
2073     }
2074
2075     return ops;
2076 }
2077
2078 /* A GFunc. */
2079 static void property_get_do_op(gpointer data,
2080                                gpointer user_data G_GNUC_UNUSED) {
2081     PropertyOp * op = data;
2082
2083     bzero(&(op->value), sizeof(op->value));
2084     op->base.result =
2085         GINT_TO_POINTER(device_property_get(op->base.child, op->id,
2086                                             &(op->value)));
2087 }
2088
2089 /* A GFunc. */
2090 static void property_set_do_op(gpointer data,
2091                                gpointer user_data G_GNUC_UNUSED) {
2092     PropertyOp * op = data;
2093
2094     op->base.result =
2095         GINT_TO_POINTER(device_property_set_ex(op->base.child, op->id,
2096                                                &(op->value), op->surety,
2097                                                op->source));
2098     g_value_unset(&(op->value));
2099 }
2100
2101 /* PropertyGetFns and PropertySetFns */
2102
2103 static gboolean
2104 property_get_block_size_fn(Device *dself,
2105     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2106     PropertySurety *surety, PropertySource *source)
2107 {
2108     RaitDevice *self = RAIT_DEVICE(dself);
2109     gsize my_block_size;
2110
2111     if (dself->block_size_source != PROPERTY_SOURCE_DEFAULT) {
2112         my_block_size = dself->block_size;
2113
2114         if (surety)
2115             *surety = dself->block_size_surety;
2116     } else {
2117         gsize child_block_size;
2118         child_block_size = calculate_block_size_from_children(self,
2119                                                     &my_block_size);
2120         if (child_block_size == 0)
2121             return FALSE;
2122
2123         if (surety)
2124             *surety = PROPERTY_SURETY_BAD; /* may still change */
2125     }
2126
2127     if (val) {
2128         g_value_unset_init(val, G_TYPE_INT);
2129         g_assert(my_block_size < G_MAXINT); /* gsize -> gint */
2130         g_value_set_int(val, (gint)my_block_size);
2131     }
2132
2133     if (source)
2134         *source = dself->block_size_source;
2135
2136     return TRUE;
2137 }
2138
2139 static gboolean
2140 property_set_block_size_fn(Device *dself,
2141     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2142     PropertySurety surety, PropertySource source)
2143 {
2144     RaitDevice *self = RAIT_DEVICE(dself);
2145     gint my_block_size = g_value_get_int(val);
2146     guint data_children;
2147
2148     find_simple_params(self, NULL, &data_children);
2149     if ((my_block_size % data_children) != 0) {
2150         device_set_error(dself,
2151             vstrallocf(_("Block size must be a multiple of %d"), data_children),
2152             DEVICE_STATUS_DEVICE_ERROR);
2153         return FALSE;
2154     }
2155
2156     dself->block_size = my_block_size;
2157     dself->block_size_source = source;
2158     dself->block_size_surety = surety;
2159
2160     if (!fix_block_size(self))
2161         return FALSE;
2162
2163     return TRUE;
2164 }
2165
2166 static gboolean
2167 property_get_canonical_name_fn(Device *dself,
2168     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2169     PropertySurety *surety, PropertySource *source)
2170 {
2171     RaitDevice *self = RAIT_DEVICE(dself);
2172     char *canonical = child_device_names_to_rait_name(self);
2173
2174     if (val) {
2175         g_value_unset_init(val, G_TYPE_STRING);
2176         g_value_set_string(val, canonical);
2177         g_free(canonical);
2178     }
2179
2180     if (surety)
2181         *surety = PROPERTY_SURETY_GOOD;
2182
2183     if (source)
2184         *source = PROPERTY_SOURCE_DETECTED;
2185
2186     return TRUE;
2187 }
2188
2189 static gboolean
2190 property_get_concurrency_fn(Device *dself,
2191     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2192     PropertySurety *surety, PropertySource *source)
2193 {
2194     RaitDevice *self = RAIT_DEVICE(dself);
2195     ConcurrencyParadigm result;
2196     guint i;
2197     GPtrArray * ops;
2198     gboolean success;
2199
2200     ops = make_property_op_array(self, PROPERTY_CONCURRENCY, NULL, 0, 0);
2201     do_rait_child_ops(self, property_get_do_op, ops);
2202
2203     /* find the most restrictive paradigm acceptable to all
2204      * child devices */
2205     result = CONCURRENCY_PARADIGM_RANDOM_ACCESS;
2206     success = TRUE;
2207     for (i = 0; i < ops->len; i ++) {
2208         ConcurrencyParadigm cur;
2209         PropertyOp * op = g_ptr_array_index(ops, i);
2210
2211         if (!op->base.result
2212             || G_VALUE_TYPE(&(op->value)) != CONCURRENCY_PARADIGM_TYPE) {
2213             success = FALSE;
2214             break;
2215         }
2216
2217         cur = g_value_get_enum(&(op->value));
2218         if (result == CONCURRENCY_PARADIGM_EXCLUSIVE ||
2219             cur == CONCURRENCY_PARADIGM_EXCLUSIVE) {
2220             result = CONCURRENCY_PARADIGM_EXCLUSIVE;
2221         } else if (result == CONCURRENCY_PARADIGM_SHARED_READ ||
2222                    cur == CONCURRENCY_PARADIGM_SHARED_READ) {
2223             result = CONCURRENCY_PARADIGM_SHARED_READ;
2224         } else if (result == CONCURRENCY_PARADIGM_RANDOM_ACCESS &&
2225                    cur == CONCURRENCY_PARADIGM_RANDOM_ACCESS) {
2226             result = CONCURRENCY_PARADIGM_RANDOM_ACCESS;
2227         } else {
2228             success = FALSE;
2229             break;
2230         }
2231     }
2232
2233     g_ptr_array_free_full(ops);
2234
2235     if (success) {
2236         if (val) {
2237             g_value_unset_init(val, CONCURRENCY_PARADIGM_TYPE);
2238             g_value_set_enum(val, result);
2239         }
2240
2241         if (surety)
2242             *surety = PROPERTY_SURETY_GOOD;
2243
2244         if (source)
2245             *source = PROPERTY_SOURCE_DETECTED;
2246     }
2247
2248     return success;
2249 }
2250
2251 static gboolean
2252 property_get_streaming_fn(Device *dself,
2253     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2254     PropertySurety *surety, PropertySource *source)
2255 {
2256     RaitDevice *self = RAIT_DEVICE(dself);
2257     StreamingRequirement result;
2258     guint i;
2259     GPtrArray * ops;
2260     gboolean success;
2261
2262     ops = make_property_op_array(self, PROPERTY_STREAMING, NULL, 0, 0);
2263     do_rait_child_ops(self, property_get_do_op, ops);
2264
2265     /* combine the child streaming requirements, selecting the strongest
2266      * requirement of the bunch. */
2267     result = STREAMING_REQUIREMENT_NONE;
2268     success = TRUE;
2269     for (i = 0; i < ops->len; i ++) {
2270         StreamingRequirement cur;
2271         PropertyOp * op = g_ptr_array_index(ops, i);
2272
2273         if (!op->base.result
2274             || G_VALUE_TYPE(&(op->value)) != STREAMING_REQUIREMENT_TYPE) {
2275             success = FALSE;
2276             break;
2277         }
2278
2279         cur = g_value_get_enum(&(op->value));
2280         if (result == STREAMING_REQUIREMENT_REQUIRED ||
2281             cur == STREAMING_REQUIREMENT_REQUIRED) {
2282             result = STREAMING_REQUIREMENT_REQUIRED;
2283         } else if (result == STREAMING_REQUIREMENT_DESIRED ||
2284                    cur == STREAMING_REQUIREMENT_DESIRED) {
2285             result = STREAMING_REQUIREMENT_DESIRED;
2286         } else if (result == STREAMING_REQUIREMENT_NONE &&
2287                    cur == STREAMING_REQUIREMENT_NONE) {
2288             result = STREAMING_REQUIREMENT_NONE;
2289         } else {
2290             success = FALSE;
2291             break;
2292         }
2293     }
2294
2295     g_ptr_array_free_full(ops);
2296
2297     if (success) {
2298         if (val) {
2299             g_value_unset_init(val, STREAMING_REQUIREMENT_TYPE);
2300             g_value_set_enum(val, result);
2301         }
2302
2303         if (surety)
2304             *surety = PROPERTY_SURETY_GOOD;
2305
2306         if (source)
2307             *source = PROPERTY_SOURCE_DETECTED;
2308     }
2309
2310     return success;
2311 }
2312
2313 static gboolean
2314 property_get_boolean_and_fn(Device *dself,
2315     DevicePropertyBase *base, GValue *val,
2316     PropertySurety *surety, PropertySource *source)
2317 {
2318     RaitDevice *self = RAIT_DEVICE(dself);
2319     gboolean result;
2320     guint i;
2321     GPtrArray * ops;
2322     gboolean success;
2323
2324     ops = make_property_op_array(self, base->ID, NULL, 0, 0);
2325     do_rait_child_ops(self, property_get_do_op, ops);
2326
2327     /* combine the child values, applying a simple AND */
2328     result = TRUE;
2329     success = TRUE;
2330     for (i = 0; i < ops->len; i ++) {
2331         PropertyOp * op = g_ptr_array_index(ops, i);
2332
2333         if (!op->base.result || !G_VALUE_HOLDS_BOOLEAN(&(op->value))) {
2334             success = FALSE;
2335             break;
2336         }
2337
2338         if (!g_value_get_boolean(&(op->value))) {
2339             result = FALSE;
2340             break;
2341         }
2342     }
2343
2344     g_ptr_array_free_full(ops);
2345
2346     if (success) {
2347         if (val) {
2348             g_value_unset_init(val, G_TYPE_BOOLEAN);
2349             g_value_set_boolean(val, result);
2350         }
2351
2352         if (surety)
2353             *surety = PROPERTY_SURETY_GOOD;
2354
2355         if (source)
2356             *source = PROPERTY_SOURCE_DETECTED;
2357     }
2358
2359     return success;
2360 }
2361
2362 static gboolean
2363 property_get_medium_access_type_fn(Device *dself,
2364     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2365     PropertySurety *surety, PropertySource *source)
2366 {
2367     RaitDevice *self = RAIT_DEVICE(dself);
2368     MediaAccessMode result;
2369     guint i;
2370     GPtrArray * ops;
2371     gboolean success;
2372
2373     ops = make_property_op_array(self, PROPERTY_MEDIUM_ACCESS_TYPE, NULL, 0, 0);
2374     do_rait_child_ops(self, property_get_do_op, ops);
2375
2376     /* combine the modes as best we can */
2377     result = 0;
2378     success = TRUE;
2379     for (i = 0; i < ops->len; i ++) {
2380         MediaAccessMode cur;
2381         PropertyOp * op = g_ptr_array_index(ops, i);
2382
2383         if (!op->base.result || G_VALUE_TYPE(&(op->value)) != MEDIA_ACCESS_MODE_TYPE) {
2384             success = FALSE;
2385             break;
2386         }
2387
2388         cur = g_value_get_enum(&(op->value));
2389
2390         if (i == 0) {
2391             result = cur;
2392         } else if ((result == MEDIA_ACCESS_MODE_READ_ONLY &&
2393                     cur == MEDIA_ACCESS_MODE_WRITE_ONLY) ||
2394                    (result == MEDIA_ACCESS_MODE_WRITE_ONLY &&
2395                     cur == MEDIA_ACCESS_MODE_READ_ONLY)) {
2396             /* Invalid combination; one device can only read, other
2397                can only write. */
2398             success = FALSE;
2399             break;
2400         } else if (result == MEDIA_ACCESS_MODE_READ_ONLY ||
2401                    cur == MEDIA_ACCESS_MODE_READ_ONLY) {
2402             result = MEDIA_ACCESS_MODE_READ_ONLY;
2403         } else if (result == MEDIA_ACCESS_MODE_WRITE_ONLY ||
2404                    cur == MEDIA_ACCESS_MODE_WRITE_ONLY) {
2405             result = MEDIA_ACCESS_MODE_WRITE_ONLY;
2406         } else if (result == MEDIA_ACCESS_MODE_WORM ||
2407                    cur == MEDIA_ACCESS_MODE_WORM) {
2408             result = MEDIA_ACCESS_MODE_WORM;
2409         } else if (result == MEDIA_ACCESS_MODE_READ_WRITE &&
2410                    cur == MEDIA_ACCESS_MODE_READ_WRITE) {
2411             result = MEDIA_ACCESS_MODE_READ_WRITE;
2412         } else {
2413             success = FALSE;
2414             break;
2415         }
2416     }
2417
2418     g_ptr_array_free_full(ops);
2419
2420     if (success) {
2421         if (val) {
2422             g_value_unset_init(val, MEDIA_ACCESS_MODE_TYPE);
2423             g_value_set_enum(val, result);
2424         }
2425
2426         if (surety)
2427             *surety = PROPERTY_SURETY_GOOD;
2428
2429         if (source)
2430             *source = PROPERTY_SOURCE_DETECTED;
2431     }
2432
2433     return success;
2434 }
2435
2436 static gboolean
2437 property_get_max_volume_usage_fn(Device *dself,
2438     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2439     PropertySurety *surety, PropertySource *source)
2440 {
2441     RaitDevice *self = RAIT_DEVICE(dself);
2442     guint64 result;
2443     guint i;
2444     GPtrArray * ops;
2445     guint data_children;
2446
2447     ops = make_property_op_array(self, PROPERTY_MAX_VOLUME_USAGE, NULL, 0, 0);
2448     do_rait_child_ops(self, property_get_do_op, ops);
2449
2450     /* look for the smallest value that is set */
2451     result = 0;
2452     for (i = 0; i < ops->len; i ++) {
2453         guint64 cur;
2454         PropertyOp * op = g_ptr_array_index(ops, i);
2455
2456         if (!op->base.result || !G_VALUE_HOLDS_UINT64(&(op->value))) {
2457             continue; /* ignore children without this property */
2458         }
2459
2460         cur = g_value_get_uint64(&(op->value));
2461
2462         if (!result || (cur && cur < result)) {
2463             result = cur;
2464         }
2465     }
2466
2467     g_ptr_array_free_full(ops);
2468
2469     if (result) {
2470         /* result contains the minimum usage on any child.  We can use that space
2471          * on each of our data children, so the total is larger */
2472         find_simple_params(self, NULL, &data_children);
2473         result *= data_children;
2474
2475         if (val) {
2476             g_value_unset_init(val, G_TYPE_UINT64);
2477             g_value_set_uint64(val, result);
2478         }
2479
2480         if (surety)
2481             *surety = PROPERTY_SURETY_GOOD;
2482
2483         if (source)
2484             *source = PROPERTY_SOURCE_DETECTED;
2485
2486         return TRUE;
2487     } else {
2488         /* no result from any children, so we effectively don't have this property */
2489         return FALSE;
2490     }
2491 }
2492
2493 static gboolean
2494 property_set_max_volume_usage_fn(Device *dself,
2495     DevicePropertyBase *base G_GNUC_UNUSED, GValue *val,
2496     PropertySurety surety, PropertySource source)
2497 {
2498     RaitDevice *self = RAIT_DEVICE(dself);
2499     guint64 parent_usage;
2500     guint64 child_usage;
2501     GValue child_val;
2502     guint i;
2503     gboolean success;
2504     GPtrArray * ops;
2505     guint data_children;
2506
2507     parent_usage = g_value_get_uint64(val);
2508     find_simple_params(self, NULL, &data_children);
2509
2510     child_usage = parent_usage / data_children;
2511
2512     bzero(&child_val, sizeof(child_val));
2513     g_value_init(&child_val, G_TYPE_UINT64);
2514     g_value_set_uint64(&child_val, child_usage);
2515
2516     ops = make_property_op_array(self, PROPERTY_MAX_VOLUME_USAGE,
2517                                 &child_val, surety, source);
2518     do_rait_child_ops(self, property_set_do_op, ops);
2519
2520     /* if any of the kids succeeded, then we did too */
2521     success = FALSE;
2522     for (i = 0; i < ops->len; i ++) {
2523         PropertyOp * op = g_ptr_array_index(ops, i);
2524
2525         if (op->base.result) {
2526             success = TRUE;
2527             break;
2528         }
2529     }
2530
2531     g_ptr_array_free_full(ops);
2532
2533     return success;
2534 }
2535
2536 typedef struct {
2537     GenericOp base;
2538     guint filenum;
2539 } RecycleFileOp;
2540
2541 /* A GFunc */
2542 static void recycle_file_do_op(gpointer data,
2543                                gpointer user_data G_GNUC_UNUSED) {
2544     RecycleFileOp * op = data;
2545     op->base.result =
2546         GINT_TO_POINTER(device_recycle_file(op->base.child, op->filenum));
2547 }
2548
2549 static gboolean
2550 rait_device_recycle_file (Device * dself, guint filenum) {
2551     GPtrArray * ops;
2552     guint i;
2553     gboolean success;
2554
2555     RaitDevice * self = RAIT_DEVICE(dself);
2556
2557     if (rait_device_in_error(self)) return FALSE;
2558
2559     ops = g_ptr_array_sized_new(self->private->children->len);
2560     for (i = 0; i < self->private->children->len; i ++) {
2561         RecycleFileOp * op;
2562         op = g_new(RecycleFileOp, 1);
2563         op->base.child = g_ptr_array_index(self->private->children, i);
2564         op->filenum = filenum;
2565         g_ptr_array_add(ops, op);
2566     }
2567
2568     do_rait_child_ops(self, recycle_file_do_op, ops);
2569
2570     success = g_ptr_array_and(ops, extract_boolean_generic_op);
2571
2572     g_ptr_array_free_full(ops);
2573
2574     if (!success) {
2575         /* TODO: be more specific here */
2576         device_set_error(dself,
2577             stralloc(_("One or more devices failed to recycle_file")),
2578             DEVICE_STATUS_DEVICE_ERROR);
2579         return FALSE;
2580     }
2581     return TRUE;
2582 }
2583
2584 /* GFunc */
2585 static void finish_do_op(gpointer data, gpointer user_data G_GNUC_UNUSED) {
2586     GenericOp * op = data;
2587     op->result = GINT_TO_POINTER(device_finish(op->child));
2588 }
2589
2590 static gboolean
2591 rait_device_finish (Device * self) {
2592     GPtrArray * ops;
2593     gboolean success;
2594     gboolean rval = TRUE;
2595
2596     rval = !rait_device_in_error(self);
2597
2598     ops = make_generic_boolean_op_array(RAIT_DEVICE(self));
2599
2600     do_rait_child_ops(RAIT_DEVICE(self), finish_do_op, ops);
2601
2602     success = g_ptr_array_and(ops, extract_boolean_generic_op);
2603     if (!success)
2604         rval = FALSE;
2605
2606     g_ptr_array_free_full(ops);
2607
2608     self->access_mode = ACCESS_NULL;
2609
2610     return rval;
2611 }
2612
2613 static Device *
2614 rait_device_factory (char * device_name, char * device_type, char * device_node) {
2615     Device * rval;
2616     g_assert(0 == strcmp(device_type, "rait"));
2617     rval = DEVICE(g_object_new(TYPE_RAIT_DEVICE, NULL));
2618     device_open_device(rval, device_name, device_type, device_node);
2619     return rval;
2620 }
2621
2622 void
2623 rait_device_register (void) {
2624     static const char * device_prefix_list[] = {"rait", NULL};
2625     register_device(rait_device_factory, device_prefix_list);
2626 }