SQUASHFS/squashfs-tools-4.4/squashfs-tools/process_fragments.c

   1 /*
   2  * Create a squashfs filesystem.  This is a highly compressed read only
   3  * filesystem.
   4  *
   5  * Copyright (c) 2014
   6  * Phillip Lougher <phillip@squashfs.org.uk>
   7  *
   8  * This program is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU General Public License
  10  * as published by the Free Software Foundation; either version 2,
  11  * or (at your option) any later version.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  21  *
  22  * process_fragments.c
  23  */
  24
  25 #include <pthread.h>
  26 #include <sys/ioctl.h>
  27 #include <unistd.h>
  28 #include <signal.h>
  29 #include <sys/time.h>
  30 #include <string.h>
  31 #include <stdio.h>
  32 #include <math.h>
  33 #include <stdarg.h>
  34 #include <errno.h>
  35 #include <stdlib.h>
  36 #include <dirent.h>
  37 #include <sys/types.h>
  38 #include <sys/stat.h>
  39 #include <fcntl.h>
  40
  41 #include "caches-queues-lists.h"
  42 #include "squashfs_fs.h"
  43 #include "mksquashfs.h"
  44 #include "error.h"
  45 #include "progressbar.h"
  46 #include "info.h"
  47 #include "compressor.h"
  48 #include "process_fragments.h"
  49
  50 #define FALSE 0
  51 #define TRUE 1
  52
  53 extern struct queue *to_process_frag;
  54 extern struct seq_queue *to_main;
  55 extern int sparse_files;
  56 extern long long start_offset;
  57
  58 /*
  59  * Compute 16 bit BSD checksum over the data, and check for sparseness
  60  */
  61 static int checksum_sparse(struct file_buffer *file_buffer)
  62 {
  63         unsigned char *b = (unsigned char *) file_buffer->data;
  64         unsigned short chksum = 0;
  65         int bytes = file_buffer->size, sparse = TRUE, value;
  66
  67         while(bytes --) {
  68                 chksum = (chksum & 1) ? (chksum >> 1) | 0x8000 : chksum >> 1;
  69                 value = *b++;
  70                 if(value) {
  71                         sparse = FALSE;
  72                         chksum += value;
  73                 }
  74         }
  75
  76         file_buffer->checksum = chksum;
  77         return sparse;
  78 }
  79
  80
  81 static int read_filesystem(int fd, long long byte, int bytes, void *buff)
  82 {
  83         off_t off = byte;
  84
  85         TRACE("read_filesystem: reading from position 0x%llx, bytes %d\n",
  86                 byte, bytes);
  87
  88         if(lseek(fd, start_offset + off, SEEK_SET) == -1) {
  89                 ERROR("read_filesystem: Lseek on destination failed because %s, "
  90                         "offset=0x%llx\n", strerror(errno), start_offset + off);
  91                 return 0;
  92         } else if(read_bytes(fd, buff, bytes) < bytes) {
  93                 ERROR("Read on destination failed\n");
  94                 return 0;
  95         }
  96
  97         return 1;
  98 }
  99
 100
 101 static struct file_buffer *get_fragment(struct fragment *fragment,
 102         char *data_buffer, int fd)
 103 {
 104         struct squashfs_fragment_entry *disk_fragment;
 105         struct file_buffer *buffer, *compressed_buffer;
 106         long long start_block;
 107         int res, size, index = fragment->index;
 108         char locked;
 109
 110         /*
 111          * Lookup fragment block in cache.
 112          * If the fragment block doesn't exist, then get the compressed version
 113          * from the writer cache or off disk, and decompress it.
 114          *
 115          * This routine has two things which complicate the code:
 116          *
 117          *      1. Multiple threads can simultaneously lookup/create the
 118          *         same buffer.  This means a buffer needs to be "locked"
 119          *         when it is being filled in, to prevent other threads from
 120          *         using it when it is not ready.  This is because we now do
 121          *         fragment duplicate checking in parallel.
 122          *      2. We have two caches which need to be checked for the
 123          *         presence of fragment blocks: the normal fragment cache
 124          *         and a "reserve" cache.  The reserve cache is used to
 125          *         prevent an unnecessary pipeline stall when the fragment cache
 126          *         is full of fragments waiting to be compressed.
 127          */
 128         pthread_cleanup_push((void *) pthread_mutex_unlock, &dup_mutex);
 129         pthread_mutex_lock(&dup_mutex);
 130
 131 again:
 132         buffer = cache_lookup_nowait(fragment_buffer, index, &locked);
 133         if(buffer) {
 134                 pthread_mutex_unlock(&dup_mutex);
 135                 if(locked)
 136                         /* got a buffer being filled in.  Wait for it */
 137                         cache_wait_unlock(buffer);
 138                 goto finished;
 139         }
 140
 141         /* not in fragment cache, is it in the reserve cache? */
 142         buffer = cache_lookup_nowait(reserve_cache, index, &locked);
 143         if(buffer) {
 144                 pthread_mutex_unlock(&dup_mutex);
 145                 if(locked)
 146                         /* got a buffer being filled in.  Wait for it */
 147                         cache_wait_unlock(buffer);
 148                 goto finished;
 149         }
 150
 151         /* in neither cache, try to get it from the fragment cache */
 152         buffer = cache_get_nowait(fragment_buffer, index);
 153         if(!buffer) {
 154                 /*
 155                  * no room, get it from the reserve cache, this is
 156                  * dimensioned so it will always have space (no more than
 157                  * processors + 1 can have an outstanding reserve buffer)
 158                  */
 159                 buffer = cache_get_nowait(reserve_cache, index);
 160                 if(!buffer) {
 161                         /* failsafe */
 162                         ERROR("no space in reserve cache\n");
 163                         goto again;
 164                 }
 165         }
 166
 167         pthread_mutex_unlock(&dup_mutex);
 168
 169         compressed_buffer = cache_lookup(fwriter_buffer, index);
 170
 171         pthread_cleanup_push((void *) pthread_mutex_unlock, &fragment_mutex);
 172         pthread_mutex_lock(&fragment_mutex);
 173         disk_fragment = &fragment_table[index];
 174         size = SQUASHFS_COMPRESSED_SIZE_BLOCK(disk_fragment->size);
 175         start_block = disk_fragment->start_block;
 176         pthread_cleanup_pop(1);
 177
 178         if(SQUASHFS_COMPRESSED_BLOCK(disk_fragment->size)) {
 179                 int error;
 180                 char *data;
 181
 182                 if(compressed_buffer)
 183                         data = compressed_buffer->data;
 184                 else {
 185                         res = read_filesystem(fd, start_block, size, data_buffer);
 186                         if(res == 0) {
 187                                 ERROR("Failed to read fragment from output"
 188                                         " filesystem\n");
 189                                 BAD_ERROR("Output filesystem corrupted?\n");
 190                         }
 191                         data = data_buffer;
 192                 }
 193
 194                 res = compressor_uncompress(comp, buffer->data, data, size,
 195                         block_size, &error);
 196                 if(res == -1)
 197                         BAD_ERROR("%s uncompress failed with error code %d\n",
 198                                 comp->name, error);
 199         } else if(compressed_buffer)
 200                 memcpy(buffer->data, compressed_buffer->data, size);
 201         else {
 202                 res = read_filesystem(fd, start_block, size, buffer->data);
 203                 if(res == 0) {
 204                         ERROR("Failed to read fragment from output "
 205                                 "filesystem\n");
 206                         BAD_ERROR("Output filesystem corrupted?\n");
 207                 }
 208         }
 209
 210         cache_unlock(buffer);
 211         cache_block_put(compressed_buffer);
 212
 213 finished:
 214         pthread_cleanup_pop(0);
 215
 216         return buffer;
 217 }
 218
 219
 220 struct file_buffer *get_fragment_cksum(struct file_info *file,
 221         char *data_buffer, int fd, unsigned short *checksum)
 222 {
 223         struct file_buffer *frag_buffer;
 224         struct append_file *append;
 225         int index = file->fragment->index;
 226
 227         frag_buffer = get_fragment(file->fragment, data_buffer, fd);
 228
 229         pthread_cleanup_push((void *) pthread_mutex_unlock, &dup_mutex);
 230
 231         for(append = file_mapping[index]; append; append = append->next) {
 232                 int offset = append->file->fragment->offset;
 233                 int size = append->file->fragment->size;
 234                 char *data = frag_buffer->data + offset;
 235                 unsigned short cksum = get_checksum_mem(data, size);
 236
 237                 if(file == append->file)
 238                         *checksum = cksum;
 239
 240                 pthread_mutex_lock(&dup_mutex);
 241                 append->file->fragment_checksum = cksum;
 242                 append->file->have_frag_checksum = TRUE;
 243                 pthread_mutex_unlock(&dup_mutex);
 244         }
 245
 246         pthread_cleanup_pop(0);
 247
 248         return frag_buffer;
 249 }
 250
 251
 252 void *frag_thrd(void *destination_file)
 253 {
 254         sigset_t sigmask, old_mask;
 255         char *data_buffer;
 256         int fd;
 257
 258         sigemptyset(&sigmask);
 259         sigaddset(&sigmask, SIGINT);
 260         sigaddset(&sigmask, SIGTERM);
 261         sigaddset(&sigmask, SIGUSR1);
 262         pthread_sigmask(SIG_BLOCK, &sigmask, &old_mask);
 263
 264         fd = open(destination_file, O_RDONLY);
 265         if(fd == -1)
 266                 BAD_ERROR("frag_thrd: can't open destination for reading\n");
 267
 268         data_buffer = malloc(SQUASHFS_FILE_MAX_SIZE);
 269         if(data_buffer == NULL)
 270                 MEM_ERROR();
 271
 272         pthread_cleanup_push((void *) pthread_mutex_unlock, &dup_mutex);
 273
 274         while(1) {
 275                 struct file_buffer *file_buffer = queue_get(to_process_frag);
 276                 struct file_buffer *buffer;
 277                 int sparse = checksum_sparse(file_buffer);
 278                 struct file_info *dupl_ptr;
 279                 long long file_size;
 280                 unsigned short checksum;
 281                 char flag;
 282                 int res;
 283
 284                 if(sparse_files && sparse) {
 285                         file_buffer->c_byte = 0;
 286                         file_buffer->fragment = FALSE;
 287                 } else
 288                         file_buffer->c_byte = file_buffer->size;
 289
 290                 /*
 291                  * Specutively pull into the fragment cache any fragment blocks
 292                  * which contain fragments which *this* fragment may be
 293                  * be a duplicate.
 294                  *
 295                  * By ensuring the fragment block is in cache ahead of time
 296                  * should eliminate the parallelisation stall when the
 297                  * main thread needs to read the fragment block to do a
 298                  * duplicate check on it.
 299                  *
 300                  * If this is a fragment belonging to a larger file
 301                  * (with additional blocks) then ignore it.  Here we're
 302                  * interested in the "low hanging fruit" of files which
 303                  * consist of only a fragment
 304                  */
 305                 if(file_buffer->file_size != file_buffer->size) {
 306                         seq_queue_put(to_main, file_buffer);
 307                         continue;
 308                 }
 309
 310                 file_size = file_buffer->file_size;
 311
 312                 pthread_mutex_lock(&dup_mutex);
 313                 dupl_ptr = dupl[DUP_HASH(file_size)];
 314                 pthread_mutex_unlock(&dup_mutex);
 315
 316                 file_buffer->dupl_start = dupl_ptr;
 317                 file_buffer->duplicate = FALSE;
 318
 319                 for(; dupl_ptr; dupl_ptr = dupl_ptr->next) {
 320                         if(file_size != dupl_ptr->file_size ||
 321                                         file_size != dupl_ptr->fragment->size)
 322                                 continue;
 323
 324                         pthread_mutex_lock(&dup_mutex);
 325                         flag = dupl_ptr->have_frag_checksum;
 326                         checksum = dupl_ptr->fragment_checksum;
 327                         pthread_mutex_unlock(&dup_mutex);
 328
 329                         /*
 330                          * If we have the checksum and it matches then
 331                          * read in the fragment block.
 332                          *
 333                          * If we *don't* have the checksum, then we are
 334                          * appending, and the fragment block is on the
 335                          * "old" filesystem.  Read it in and checksum
 336                          * the entire fragment buffer
 337                          */
 338                         if(!flag) {
 339                                 buffer = get_fragment_cksum(dupl_ptr,
 340                                         data_buffer, fd, &checksum);
 341                                 if(checksum != file_buffer->checksum) {
 342                                         cache_block_put(buffer);
 343                                         continue;
 344                                 }
 345                         } else if(checksum == file_buffer->checksum)
 346                                 buffer = get_fragment(dupl_ptr->fragment,
 347                                         data_buffer, fd);
 348                         else
 349                                 continue;
 350
 351                         res = memcmp(file_buffer->data, buffer->data +
 352                                 dupl_ptr->fragment->offset, file_size);
 353                         cache_block_put(buffer);
 354                         if(res == 0) {
 355                                 struct file_buffer *dup = malloc(sizeof(*dup));
 356                                 if(dup == NULL)
 357                                         MEM_ERROR();
 358                                 memcpy(dup, file_buffer, sizeof(*dup));
 359                                 cache_block_put(file_buffer);
 360                                 dup->dupl_start = dupl_ptr;
 361                                 dup->duplicate = TRUE;
 362                                 file_buffer = dup;
 363                                 break;
 364                         }
 365                 }
 366
 367                 seq_queue_put(to_main, file_buffer);
 368         }
 369
 370         pthread_cleanup_pop(0);
 371 }