taper-disk-port-source.c

开源备份软件源码 AMANDA, the Advanced Maryland Automatic Network Disk Archiver, is a backup system that a

                               selfp->disk_buffered_bytes);
}

/* Handle the output of the small amount of saved in-memory data. */
static size_t handle_excess_buffer_read(TaperDiskPortSource * self,
                                        void * buf, size_t count) {
    TaperSource * pself = (TaperSource*)self;
    guint64 offset;

    /* First, do we have anything left? */
    if (selfp->retry_data_written >=
        (selfp->disk_buffered_bytes + selfp->excess_buffer_size)) {
        pself->end_of_part = TRUE;
        return 0;
    }
    
    count = MIN(count,
                (selfp->disk_buffered_bytes + selfp->excess_buffer_size)
                - selfp->retry_data_written);

    offset = selfp->disk_buffered_bytes + selfp->excess_buffer_size
        - selfp->retry_data_written;
    g_assert(offset + count <= selfp->excess_buffer_size);
    memcpy(buf, selfp->excess_buffer + offset, count);

    selfp->retry_data_written += count;

    return count;
}
    
/* Write data out to the disk buffer, and handle any problems that
   crop up along the way. */
static ssize_t write_disk_buffer(TaperDiskPortSource * self, char * buf,
                                 size_t read_size) {
    size_t bytes_written = 0;
    while (bytes_written < read_size) {
        int write_result = write(selfp->buffer_fd, buf + bytes_written,
                                 read_size - bytes_written);
        if (write_result > 0) {
            bytes_written += write_result;
            continue;
        } else if (write_result == 0) {
            g_fprintf(stderr, "Writing disk buffer: Wrote 0 bytes.\n");
            continue;
        } else {
            if (0
#ifdef EAGAIN
                || errno == EAGAIN
#endif
#ifdef EWOULDBLOCK
                || errno == EWOULDBLOCK
#endif
#ifdef EINTR
                || errno == EINTR
#endif
                ) {
                /* Try again. */
                continue;
            } else if (0
#ifdef EFBIG
                       || errno == EFBIG
#endif
#ifdef ENOSPC
                       || errno == ENOSPC
#endif
                       ) {
                /* Out of space */
                store_excess(self, buf, read_size, bytes_written);
                return read_size;
            } else {
                /* I/O error. */
                store_excess(self, buf, read_size, bytes_written);
                selfp->disk_problem = TRUE;
                TAPER_SOURCE(self)->end_of_part = TRUE;
                return read_size;
            }
        }
        g_assert_not_reached();
    }

    selfp->disk_buffered_bytes += bytes_written;
    return read_size;
}

static ssize_t 
taper_disk_port_source_read (TaperSource * pself, void * buf, size_t count) {
    TaperDiskPortSource * self = (TaperDiskPortSource*)pself;

    g_return_val_if_fail (self != NULL, -1);
    g_return_val_if_fail (TAPER_IS_DISK_PORT_SOURCE (self), -1);
    g_return_val_if_fail (buf != NULL, -1);
    g_return_val_if_fail (count > 0, -1);
    g_assert(selfp->disk_buffered_bytes <= pself->max_part_size);
	
    if (selfp->fallback != NULL) {
        return taper_source_read(selfp->fallback, buf, count);
    } else if (selfp->buffer_fd < 0) {
        if (!open_buffer_file(self)) {
            /* Buffer file failed; go immediately to failover mode. */
            selfp->fallback = make_fallback_source(self);
            if (selfp->fallback != NULL) {
                return taper_source_read(selfp->fallback, buf, count);
            } else {
                /* Even the fallback source failed! */
                return -1;
            }
        }
    }

    if (selfp->retry_mode) {
        /* Read from disk buffer. */

        if (selfp->retry_data_written < selfp->disk_buffered_bytes) {
            /* Read from disk. */
            int result;
            count = MIN(count, selfp->disk_buffered_bytes -
                               selfp->retry_data_written);
            result = read(selfp->buffer_fd, buf, count);
            if (result <= 0) {
                /* This should not happen. */
                return -1;
            } else {
                selfp->retry_data_written += result;
                return result;
            }
        } else if (selfp->excess_buffer != NULL) {
            /* We are writing out the last bit of buffer. Handle that. */
            return handle_excess_buffer_read(self, buf, count);
        } else {
            /* No more data. */
            pself->end_of_part = TRUE;
            return 0;
        }
        
        g_assert_not_reached();
    } else {
        /* Read from port. */
        int read_result;
        count = MIN(count, pself->max_part_size - selfp->disk_buffered_bytes);
        if (count == 0) /* It was nonzero before. */ {
            pself->end_of_part = TRUE;
            return 0;
        }
        
        read_result = source_parent_class->read(pself, buf, count);
        /* Parent handles EOF and other goodness. */
        if (read_result <= 0) {
            return read_result;
        }
        /* Now write to disk buffer. */
        return write_disk_buffer(self, buf, read_result);
    }
}

/* Try seeking back to byte 0. If that fails, then we mark ourselves
   as having a disk problem. Returns FALSE in that case. */
static gboolean try_rewind(TaperDiskPortSource * self) {
    gint64 result;
    result = lseek(selfp->buffer_fd, 0, SEEK_SET);
    if (result != 0) {
        g_fprintf(stderr, "Couldn't seek split buffer: %s\n", strerror(errno));
        selfp->disk_problem = TRUE;
        return FALSE;
    } else {
        return TRUE;
    }
}

static gboolean 
taper_disk_port_source_seek_to_part_start (TaperSource * pself) {
    TaperDiskPortSource * self = (TaperDiskPortSource*)pself;
    g_return_val_if_fail (self != NULL, FALSE);
    g_return_val_if_fail (TAPER_IS_DISK_PORT_SOURCE (pself), FALSE);
    g_return_val_if_fail (selfp->disk_buffered_bytes
                          + selfp->excess_buffer_size > 0, FALSE);
    
    if (self->_priv->fallback != NULL) {
        return taper_source_seek_to_part_start(selfp->fallback);
    }

    if (selfp->disk_problem && selfp->disk_buffered_bytes) {
        /* The disk buffer is screwed; nothing to do. */
        return FALSE;
    }

    if (!selfp->disk_problem) {
        if (!try_rewind(self)) {
            return FALSE;
        }
    }

    selfp->retry_mode = TRUE;
    selfp->retry_data_written = 0;

    if (source_parent_class->seek_to_part_start) {
        return source_parent_class->seek_to_part_start(pself);
    } else {
        return TRUE;
    }
}

static void 
taper_disk_port_source_start_new_part (TaperSource * pself) {
    TaperDiskPortSource * self = (TaperDiskPortSource*)pself;
    g_return_if_fail (self != NULL);
    g_return_if_fail (TAPER_IS_DISK_PORT_SOURCE (pself));
	
    if (self->_priv->fallback != NULL) {
        taper_source_start_new_part(self->_priv->fallback);
        return;
    }

    selfp->retry_mode = FALSE;
    if (!selfp->disk_problem) {
        try_rewind(self); /* If this fails it will set disk_problem to
                             TRUE. */
    }

    if (selfp->disk_problem && selfp->fallback == NULL) {
        selfp->fallback = make_fallback_source(self);
    }
    selfp->disk_buffered_bytes = 0;
    amfree(selfp->excess_buffer);
    selfp->excess_buffer_size = selfp->retry_data_written = 0;

    if (source_parent_class->start_new_part) {
        source_parent_class->start_new_part(pself);
    }
}