io: enhance istream/bstreams with pipe to forked child

* prunes the child pid to avoid zombies
 * handles the errors so e.g. file-not-found is reported properly

src/apk_blob.h

@@ -18,7 +18,7 @@
 #include "apk_defines.h"
 
 struct apk_blob {
-	unsigned int len;
+	long len;
 	char *ptr;
 };
 typedef struct apk_blob apk_blob_t;
@@ -56,6 +56,7 @@ static inline const EVP_MD *apk_checksum_default(void)
 #define APK_BLOB_IS_NULL(blob)		((blob).ptr == NULL)
 
 #define APK_BLOB_NULL			((apk_blob_t){0, NULL})
+#define APK_BLOB_ERROR(err)		((apk_blob_t){err, NULL})
 #define APK_BLOB_BUF(buf)		((apk_blob_t){sizeof(buf), (char *)(buf)})
 #define APK_BLOB_CSUM(csum)		((apk_blob_t){(csum).type, (char *)(csum).data})
 #define APK_BLOB_STRUCT(s)		((apk_blob_t){sizeof(s), (char*)&(s)})

src/apk_io.h

@@ -73,7 +73,7 @@ static inline struct apk_istream *apk_bstream_gunzip(struct apk_bstream *bs)
 struct apk_ostream *apk_ostream_gzip(struct apk_ostream *);
 struct apk_ostream *apk_ostream_counter(off_t *);
 
-struct apk_istream *apk_istream_from_fd(int fd);
+struct apk_istream *apk_istream_from_fd_pid(int fd, pid_t pid, int (*translate_status)(int));
 struct apk_istream *apk_istream_from_file(int atfd, const char *file);
 struct apk_istream *apk_istream_from_file_gz(int atfd, const char *file);
 struct apk_istream *apk_istream_from_url(const char *url);
@@ -82,12 +82,22 @@ size_t apk_istream_skip(struct apk_istream *istream, size_t size);
 size_t apk_istream_splice(void *stream, int fd, size_t size,
 			  apk_progress_cb cb, void *cb_ctx);
 
+static inline struct apk_istream *apk_istream_from_fd(int fd)
+{
+	return apk_istream_from_fd_pid(fd, 0, NULL);
+}
+
 struct apk_bstream *apk_bstream_from_istream(struct apk_istream *istream);
-struct apk_bstream *apk_bstream_from_fd(int fd);
+struct apk_bstream *apk_bstream_from_fd_pid(int fd, pid_t pid, int (*translate_status)(int));
 struct apk_bstream *apk_bstream_from_file(int atfd, const char *file);
 struct apk_bstream *apk_bstream_from_url(const char *url);
 struct apk_bstream *apk_bstream_tee(struct apk_bstream *from, int atfd, const char *to);
 
+static inline struct apk_bstream *apk_bstream_from_fd(int fd)
+{
+	return apk_bstream_from_fd_pid(fd, 0, NULL);
+}
+
 struct apk_ostream *apk_ostream_to_fd(int fd);
 struct apk_ostream *apk_ostream_to_file(int atfd, const char *file, const char *tmpfile, mode_t mode);
 struct apk_ostream *apk_ostream_to_file_gz(int atfd, const char *file, const char *tmpfile, mode_t mode);

src/gunzip.c

@@ -61,8 +61,8 @@ static ssize_t gzi_read(void *stream, void *ptr, size_t size)
 			gis->cbprev = blob.ptr;
 			gis->zs.avail_in = blob.len;
 			gis->zs.next_in = (void *) gis->cbprev;
-			if (gis->zs.avail_in < 0) {
-				gis->err = -EIO;
+			if (blob.len < 0) {
+				gis->err = blob.len;
 				goto ret;
 			} else if (gis->zs.avail_in == 0) {
 				gis->err = 1;

src/io.c

@@ -15,6 +15,7 @@
 #include <unistd.h>
 #include <malloc.h>
 #include <sys/mman.h>
+#include <sys/wait.h>
 #include <sys/stat.h>
 #include <pwd.h>
 #include <grp.h>
@@ -26,6 +27,8 @@
 struct apk_fd_istream {
 	struct apk_istream is;
 	int fd;
+	pid_t pid;
+	int (*translate_status)(int status);
 };
 
 static ssize_t fdi_read(void *stream, void *ptr, size_t size)
@@ -45,9 +48,15 @@ static ssize_t fdi_read(void *stream, void *ptr, size_t size)
 		if (r < 0)
 			return -errno;
 		if (r == 0)
-			return i;
+			break;
 		i += r;
 	}
+	if (i == 0 && fis->pid != 0) {
+		int status;
+		if (waitpid(fis->pid, &status, 0) == fis->pid)
+			i = fis->translate_status(status);
+		fis->pid = 0;
+	}
 
 	return i;
 }
@@ -56,12 +65,15 @@ static void fdi_close(void *stream)
 {
 	struct apk_fd_istream *fis =
 		container_of(stream, struct apk_fd_istream, is);
+	int status;
 
 	close(fis->fd);
+	if (fis->pid != 0)
+		waitpid(fis->pid, &status, 0);
 	free(fis);
 }
 
-struct apk_istream *apk_istream_from_fd(int fd)
+struct apk_istream *apk_istream_from_fd_pid(int fd, pid_t pid, int (*translate_status)(int))
 {
 	struct apk_fd_istream *fis;
 
@@ -78,6 +90,8 @@ struct apk_istream *apk_istream_from_fd(int fd)
 		.is.read = fdi_read,
 		.is.close = fdi_close,
 		.fd = fd,
+		.pid = pid,
+		.translate_status = translate_status,
 	};
 
 	return &fis->is;
@@ -196,9 +210,9 @@ static apk_blob_t is_bs_read(void *stream, apk_blob_t token)
 			goto ret_all;
 	}
 
-	/* If we've exchausted earlier, it's end of stream */
+	/* If we've exchausted earlier, it's end of stream or error */
 	if (APK_BLOB_IS_NULL(isbs->left))
-		return APK_BLOB_NULL;
+		return isbs->left;
 
 	/* We need more data */
 	if (isbs->left.len != 0)
@@ -213,6 +227,10 @@ static apk_blob_t is_bs_read(void *stream, apk_blob_t token)
 		if (isbs->left.len == 0)
 			isbs->left = APK_BLOB_NULL;
 		goto ret_all;
+	} else {
+		/* cache and return error */
+		isbs->left = ret = APK_BLOB_ERROR(size);
+		goto ret;
 	}
 
 	if (!APK_BLOB_IS_NULL(token)) {
@@ -294,7 +312,6 @@ static void mmap_close(void *stream, size_t *size)
 
 	if (size != NULL)
 		*size = mbs->size;
-
 	munmap(mbs->ptr, mbs->size);
 	close(mbs->fd);
 	free(mbs);
@@ -332,18 +349,20 @@ static struct apk_bstream *apk_mmap_bstream_from_fd(int fd)
 	return &mbs->bs;
 }
 
-struct apk_bstream *apk_bstream_from_fd(int fd)
+struct apk_bstream *apk_bstream_from_fd_pid(int fd, pid_t pid, int (*translate_status)(int))
 {
 	struct apk_bstream *bs;
 
 	if (fd < 0)
 		return NULL;
 
-	bs = apk_mmap_bstream_from_fd(fd);
-	if (bs != NULL)
-		return bs;
+	if (pid == 0) {
+		bs = apk_mmap_bstream_from_fd(fd);
+		if (bs != NULL)
+			return bs;
+	}
 
-	return apk_bstream_from_istream(apk_istream_from_fd(fd));
+	return apk_bstream_from_istream(apk_istream_from_fd_pid(fd, pid, translate_status));
 }
 
 struct apk_bstream *apk_bstream_from_file(int atfd, const char *file)

src/print.c

@@ -26,7 +26,7 @@ int apk_print_indented(struct apk_indent *i, apk_blob_t blob)
 		i->x = i->indent;
 		printf("\n%*s", i->indent - 1, "");
 	}
-	i->x += printf(" %.*s", blob.len, blob.ptr);
+	i->x += printf(" %.*s", (int) blob.len, blob.ptr);
 	return 0;
 }
 

src/url.c

@@ -31,7 +31,26 @@ const char *apk_url_local_file(const char *url)
 	return NULL;
 }
 
-static int fork_wget(const char *url)
+static int translate_wget(int status)
+{
+	if (!WIFEXITED(status))
+		return -EFAULT;
+
+	switch (WEXITSTATUS(status)) {
+	case 0:
+		return 0;
+	case 3:
+		return -EIO;
+	case 4:
+		return -ECONNABORTED;
+	case 8:
+		return -ENOENT;
+	default:
+		return -EFAULT;
+	}
+}
+
+static int fork_wget(const char *url, pid_t *ppid)
 {
 	pid_t pid;
 	int fds[2];
@@ -56,15 +75,23 @@ static int fork_wget(const char *url)
 	}
 
 	close(fds[1]);
+
+	if (ppid != NULL)
+		*ppid = pid;
+
 	return fds[0];
 }
 
 struct apk_istream *apk_istream_from_url(const char *url)
 {
+	pid_t pid;
+	int fd;
+
 	if (apk_url_local_file(url) != NULL)
 		return apk_istream_from_file(AT_FDCWD, apk_url_local_file(url));
 
-	return apk_istream_from_fd(fork_wget(url));
+	fd = fork_wget(url, &pid);
+	return apk_istream_from_fd_pid(fd, pid, translate_wget);
 }
 
 struct apk_istream *apk_istream_from_url_gz(const char *file)
@@ -74,10 +101,14 @@ struct apk_istream *apk_istream_from_url_gz(const char *file)
 
 struct apk_bstream *apk_bstream_from_url(const char *url)
 {
+	pid_t pid;
+	int fd;
+
 	if (apk_url_local_file(url))
 		return apk_bstream_from_file(AT_FDCWD, url);
 
-	return apk_bstream_from_fd(fork_wget(url));
+	fd = fork_wget(url, &pid);
+	return apk_bstream_from_fd_pid(fd, pid, translate_wget);
 }
 
 int apk_url_download(const char *url, int atfd, const char *file)
@@ -102,9 +133,10 @@ int apk_url_download(const char *url, int atfd, const char *file)
 	}
 
 	waitpid(pid, &status, 0);
-	if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
+	status = translate_wget(status);
+	if (status != 0) {
 		unlinkat(atfd, file, 0);
-		return -1;
+		return status;
 	}
 
 	return 0;

src/ver.c

@@ -33,7 +33,7 @@ static int ver_indexes(struct apk_database *db, int argc, char **argv)
 			continue;
 
 		printf("%.*s [%s]\n",
-                       repo->description.len,
+                       (int) repo->description.len,
 		       repo->description.ptr,
 		       db->repos[i].url);
 	}