Merge pull request #38011 from ClickHouse/follow-up-compressor

Follow up on self-extracting-executable

utils/CMakeLists.txt

@@ -32,7 +32,7 @@ if (NOT DEFINED ENABLE_UTILS OR ENABLE_UTILS)
     add_subdirectory (check-mysql-binlog)
     add_subdirectory (keeper-bench)
     add_subdirectory (graphite-rollup)
-    add_subdirectory (self-extr-exec)
+    add_subdirectory (self-extracting-executable)
 
     if (TARGET ch_contrib::nuraft)
         add_subdirectory (keeper-data-dumper)

utils/self-extracting-executable/CMakeLists.txt

@@ -11,4 +11,5 @@ add_custom_command (TARGET compressor
     POST_BUILD
     COMMAND cat pre_compressor decompressor > compressor
     COMMAND bash ${CMAKE_CURRENT_SOURCE_DIR}/post_build.sh >> compressor
+    COMMAND chmod +x compressor
 )

utils/self-extracting-executable/compressor.cpp

@@ -11,6 +11,41 @@
 
 #include "types.h"
 
+/// blocking write
+ssize_t write_data(int fd, const void *buf, size_t count)
+{
+    for (size_t n = 0; n < count;)
+    {
+        ssize_t sz = write(fd, reinterpret_cast<const char*>(buf) + n, count - n);
+        if (sz < 0)
+        {
+            if (errno == EINTR)
+                continue;
+            return sz;
+        }
+        n += sz;
+    }
+    return count;
+}
+
+/// blocking read
+ssize_t read_data(int fd, void *buf, size_t count)
+{
+    for (size_t n = 0; n < count;)
+    {
+        ssize_t sz = read(fd, reinterpret_cast<char*>(buf) + n, count - n);
+        if (sz < 0)
+        {
+            if (errno == EINTR)
+                continue;
+            return sz;
+        }
+        if (sz == 0)
+            return count - n;
+        n += sz;
+    }
+    return count;
+}
 
 /// Main compression part
 int doCompress(char * input, char * output, off_t & in_offset, off_t & out_offset,
@@ -110,7 +145,7 @@ int compress(int in_fd, int out_fd, int level, off_t & pointer, const struct sta
         }
 
         /// Save data into file and refresh pointer
-        if (current_block_size != write(out_fd, output, current_block_size))
+        if (current_block_size != write_data(out_fd, output, current_block_size))
         {
             perror(nullptr);
             return 1;
@@ -219,7 +254,7 @@ int compressFiles(char* filenames[], int count, int output_fd, int level, const
             continue;
         }
 
-        printf("Size: %lld\n", info_in.st_size);
+        printf("Size: %td\n", info_in.st_size);
 
         /// Save umask
         files_data[i].umask = info_in.st_mode;
@@ -271,30 +306,28 @@ int copy_decompressor(const char *self, int output_fd)
 
     if (-1 == lseek(input_fd, -15, SEEK_END))
     {
-        close(input_fd);
         perror(nullptr);
+        close(input_fd);
         return 1;
     }
 
     char size_str[16] = {0};
-    for (size_t s_sz = sizeof(size_str) - 1; s_sz;)
+    if (ssize_t sz = read_data(input_fd, size_str, 15); sz < 15)
     {
-        ssize_t sz = read(input_fd, size_str + sizeof(size_str) - (s_sz + 1), s_sz);
-        if (sz <= 0)
-        {
-            close(input_fd);
+        if (sz < 0)
             perror(nullptr);
-            return 1;
-        }
-        s_sz -= sz;
+        else
+            fprintf(stderr, "Error: unable to extract decompressor.\n");
+        close(input_fd);
+        return 1;
     }
 
     int decompressor_size = atoi(size_str);
 
     if (-1 == lseek(input_fd, -(decompressor_size + 15), SEEK_END))
     {
-        close(input_fd);
         perror(nullptr);
+        close(input_fd);
         return 1;
     }
 
@@ -310,21 +343,18 @@ int copy_decompressor(const char *self, int output_fd)
 
         if (n < 0)
         {
-            close(input_fd);
+            if (errno == EINTR)
+                continue;
             perror(nullptr);
+            close(input_fd);
             return 1;
         }
 
-        while (n > 0)
+        if (n != write_data(output_fd, buf, n))
         {
-            ssize_t sz = write(output_fd, buf, n);
-            if (sz < 0)
-            {
-                close(input_fd);
-                perror(nullptr);
-                return 1;
-            }
-            n -= sz;
+            perror(nullptr);
+            close(input_fd);
+            return 1;
         }
     } while (true);
 

utils/self-extracting-executable/decompressor.cpp

@@ -1,6 +1,3 @@
-//#include <cstddef>
-//#include <cstdio>
-//#include <cstring>
 #include <zstd.h>
 #include <sys/mman.h>
 #if defined __APPLE__
@@ -55,7 +52,7 @@ int decompress(char * input, char * output, off_t start, off_t end, size_t max_n
         size = ZSTD_findFrameCompressedSize(input + in_pointer, max_block_size);
         if (ZSTD_isError(size))
         {
-            fprintf(stderr, "Error (ZSTD): %lld %s\n", size, ZSTD_getErrorName(size));
+            fprintf(stderr, "Error (ZSTD): %td %s\n", size, ZSTD_getErrorName(size));
             error_happened = true;
             break;
         }
@@ -63,7 +60,7 @@ int decompress(char * input, char * output, off_t start, off_t end, size_t max_n
         decompressed_size = ZSTD_getFrameContentSize(input + in_pointer, max_block_size);
         if (ZSTD_isError(decompressed_size))
         {
-            fprintf(stderr, "Error (ZSTD): %lld %s\n", decompressed_size, ZSTD_getErrorName(decompressed_size));
+            fprintf(stderr, "Error (ZSTD): %td %s\n", decompressed_size, ZSTD_getErrorName(decompressed_size));
             error_happened = true;
             break;
         }
@@ -174,7 +171,7 @@ int decompressFiles(int input_fd, char * path, char * name, bool & have_compress
     }
     if (fs_info.f_blocks * info_in.st_blksize < decompressed_full_size)
     {
-        fprintf(stderr, "Not enough space for decompression. Have %llu, need %zu.",
+        fprintf(stderr, "Not enough space for decompression. Have %tu, need %zu.",
                 fs_info.f_blocks * info_in.st_blksize, decompressed_full_size);
         return 1;
     }