linux - 多进程同步数据文件传递，使用select |投票 |其他？

Question

我有一个大数据文件，该文件会不断地(同步地)被现场的测量设备附加到该文件中。我需要将最新的数据同步传送到这个文件到在线仪表板。我说等时是因为“仪表板”不关心显示数据流(高延迟情况)，它只关心发送给它的文件中的最后几个数据点。我无法保证文件增长率低于我的有效出站吞吐量。

所以我有一个 不断追加的文件，但我有多个进程需要定期发送最后一个信息 block 。我猜是某种松散的 pub-sub。

我可以:

1. 轮询文件以查看是否有增长，然后从 EOF 中寻找最后的数据 block ，
2. select() 样式并收到上述更改的通知，但我是否不必查找最后一位数据？
3. 让一个进程将最后一位数据推送到共享内存中以供读取，但我会不会遇到我需要在 #1 和 #2 中解决的相同问题，因为共享内存写入器的行为相同。

还有其他建议或建议吗？

Answer 1

如果我没理解错的话，您只是想让仪表板定期更新上一个(当前) block 。那么，一个简单的选择是:

• sleep()、stat() 文件以查看它是否已更改，如果合适则发送数据。

在 Linux 上，您可以使用 inotify 在文件更改时收到通知。这样就可以避免之前方法中不必要的唤醒。所以这个选项是:

• 等待通知，如果通知到达 stat()，如果文件已更改，则发送数据和 sleep() 以避免过于频繁的更新。

最后一个可能看起来像:

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>

#include <unistd.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>

#include <sys/inotify.h>


#define BLOCK_SIZE 5


static ssize_t read_fully(int fd, void *buf, size_t count)
{
    ssize_t ret;
    size_t nread;

    nread = 0;
    do {
            ret = read(fd, buf, count);
            if (ret > 0)
                    nread += ret;
    } while ((ret > 0 && nread < count) || (ret < 0 && errno == EINTR));

    return ret < 0 ? ret : (ssize_t) nread;
}

static void show_current_block(int fd)
{
    static off_t size = 0;  /* non-theadsafe, move fd and size to a
                               context struct passed as an argument
                               if you want thread-safety            */
    signed char block[BLOCK_SIZE];  /* Assume only ASCII-compatible encoding */
    struct stat st;
    ssize_t ret;

    if (fstat(fd, &st) < 0) {
            perror("fstat");
            exit(1);
    }

    /* Handle truncated file */
    if (st.st_size < size) {
            size = 0;
    }

    if (st.st_size >= size + BLOCK_SIZE) {
            size = st.st_size / BLOCK_SIZE * BLOCK_SIZE;

            if (lseek(fd, -BLOCK_SIZE, SEEK_END) < 0) {
                    perror("lseek");
                    exit(1);
            }

            ret = read_fully(fd, block, BLOCK_SIZE);

            if (ret < 0) {
                    perror("read");
                    exit(1);
            }

            if (ret == 0) {
                    fprintf(stderr, "file closed!");
                    exit(1);
            }

            /* Assume only ASCII-compatible encoding, don't print
             * neither C0 control chars, nor > 0x7f chars (including C1)
             */
            printf("Current block: %c%c%c%c%c\n",
                    block[0] < 20 ? '.' : block[0],
                    block[1] < 20 ? '.' : block[1],
                    block[2] < 20 ? '.' : block[2],
                    block[3] < 20 ? '.' : block[3],
                    block[4] < 20 ? '.' : block[4]);

            /* Don't update too often */
            usleep(3000 * 1000);
    }
}

int main(void)
{
    int fd, ifd, wd;
    struct inotify_event ev;
    ssize_t ret;

    fd = open("testfile", O_RDONLY);
    if (fd < 0) {
            perror("open");
            exit(1);
    }

    ifd = inotify_init();
    if (ifd < 0) {
            perror("inotify_init");
            exit(1);
    }

    /* XXX race between open and inotify_add_watch */
    wd = inotify_add_watch(ifd, "testfile", IN_MODIFY);
    if (wd < 0) {
            perror("inotify_add_watch");
            exit(1);
    }

    show_current_block(fd);

    while ((ret = read(ifd, &ev, sizeof(struct inotify_event)))) {
            if (ret < 0) {
                    perror("read inotify watch");
                    exit(1);
            }
            if (ret == 0) {
                    fprintf(stderr, "inotify watch closed!\n");
                    exit(1);
            }
            if (ret != sizeof(struct inotify_event)) {
                    fprintf(stderr, "bad inotify event size %d (expected %d)\n",
                            ret, sizeof(struct inotify_event));
                    exit(1);
            }

            show_current_block(fd);
    }

    return 0;
}