diff options
| author | Mitja Felicijan <mitja.felicijan@gmail.com> | 2026-01-21 22:40:55 +0100 |
|---|---|---|
| committer | Mitja Felicijan <mitja.felicijan@gmail.com> | 2026-01-21 22:40:55 +0100 |
| commit | 5d8dfe892a2ea89f706ee140c3bdcfd89fe03fda (patch) | |
| tree | 1acdfa5220cd13b7be43a2a01368e80d306473ca /examples/redis-unstable/src/rio.h | |
| parent | c7ab12bba64d9c20ccd79b132dac475f7bc3923e (diff) | |
| download | crep-5d8dfe892a2ea89f706ee140c3bdcfd89fe03fda.tar.gz | |
Add Redis source code for testing
Diffstat (limited to 'examples/redis-unstable/src/rio.h')
| -rw-r--r-- | examples/redis-unstable/src/rio.h | 188 |
1 files changed, 188 insertions, 0 deletions
diff --git a/examples/redis-unstable/src/rio.h b/examples/redis-unstable/src/rio.h new file mode 100644 index 0000000..9cd24cd --- /dev/null +++ b/examples/redis-unstable/src/rio.h @@ -0,0 +1,188 @@ +/* + * Copyright (c) 2009-Present, Redis Ltd. + * All rights reserved. + * + * Copyright (c) 2024-present, Valkey contributors. + * All rights reserved. + * + * Licensed under your choice of (a) the Redis Source Available License 2.0 + * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the + * GNU Affero General Public License v3 (AGPLv3). + * + * Portions of this file are available under BSD3 terms; see REDISCONTRIBUTIONS for more information. + */ + + +#ifndef __REDIS_RIO_H +#define __REDIS_RIO_H + +#include <stdio.h> +#include <stdint.h> +#include "sds.h" +#include "connection.h" + +#define RIO_FLAG_READ_ERROR (1<<0) +#define RIO_FLAG_WRITE_ERROR (1<<1) + +#define RIO_TYPE_FILE (1<<0) +#define RIO_TYPE_BUFFER (1<<1) +#define RIO_TYPE_CONN (1<<2) +#define RIO_TYPE_FD (1<<3) + +struct _rio { + /* Backend functions. + * Since this functions do not tolerate short writes or reads the return + * value is simplified to: zero on error, non zero on complete success. */ + size_t (*read)(struct _rio *, void *buf, size_t len); + size_t (*write)(struct _rio *, const void *buf, size_t len); + off_t (*tell)(struct _rio *); + int (*flush)(struct _rio *); + /* The update_cksum method if not NULL is used to compute the checksum of + * all the data that was read or written so far. The method should be + * designed so that can be called with the current checksum, and the buf + * and len fields pointing to the new block of data to add to the checksum + * computation. */ + void (*update_cksum)(struct _rio *, const void *buf, size_t len); + + /* The current checksum and flags (see RIO_FLAG_*) */ + uint64_t cksum, flags; + + /* number of bytes read or written */ + size_t processed_bytes; + + /* maximum single read or write chunk size */ + size_t max_processing_chunk; + + /* Backend-specific vars. */ + union { + /* In-memory buffer target. */ + struct { + sds ptr; + off_t pos; + } buffer; + /* Stdio file pointer target. */ + struct { + FILE *fp; + off_t buffered; /* Bytes written since last fsync. */ + off_t autosync; /* fsync after 'autosync' bytes written. */ + unsigned reclaim_cache:1; /* A flag to indicate reclaim cache after fsync */ + } file; + /* Connection object (used to read from socket) */ + struct { + connection *conn; /* Connection */ + off_t pos; /* pos in buf that was returned */ + sds buf; /* buffered data */ + size_t read_limit; /* don't allow to buffer/read more than that */ + size_t read_so_far; /* amount of data read from the rio (not buffered) */ + } conn; + /* FD target (used to write to pipe). */ + struct { + int fd; /* File descriptor. */ + off_t pos; + sds buf; + } fd; + /* Multiple connections target (used to write to N sockets). */ + struct { + struct { + connection *conn; /* Connection */ + int failed; /* If write failed on this connection. */ + } *dst; + + size_t n_dst; /* Number of connections */ + off_t pos; /* Number of sent bytes */ + sds buf; + } connset; + } io; +}; + +typedef struct _rio rio; + +/* The following functions are our interface with the stream. They'll call the + * actual implementation of read / write / tell, and will update the checksum + * if needed. */ + +static inline size_t rioWrite(rio *r, const void *buf, size_t len) { + if (r->flags & (RIO_FLAG_WRITE_ERROR)) return 0; + while (len) { + size_t bytes_to_write = (r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len; + if (r->update_cksum) r->update_cksum(r,buf,bytes_to_write); + if (r->write(r,buf,bytes_to_write) == 0) { + r->flags |= RIO_FLAG_WRITE_ERROR; + return 0; + } + buf = (char*)buf + bytes_to_write; + len -= bytes_to_write; + r->processed_bytes += bytes_to_write; + } + return 1; +} + +static inline size_t rioRead(rio *r, void *buf, size_t len) { + if (r->flags & (RIO_FLAG_READ_ERROR)) return 0; + while (len) { + size_t bytes_to_read = (r->max_processing_chunk && r->max_processing_chunk < len) ? r->max_processing_chunk : len; + if (r->read(r,buf,bytes_to_read) == 0) { + r->flags |= RIO_FLAG_READ_ERROR; + return 0; + } + if (r->update_cksum) r->update_cksum(r,buf,bytes_to_read); + buf = (char*)buf + bytes_to_read; + len -= bytes_to_read; + r->processed_bytes += bytes_to_read; + } + return 1; +} + +static inline off_t rioTell(rio *r) { + return r->tell(r); +} + +static inline int rioFlush(rio *r) { + return r->flush(r); +} + +/* Abort RIO asynchronously by setting read and write error flags. Subsequent + * rioRead()/rioWrite() calls will fail, letting the caller terminate safely. */ +static inline void rioAbort(rio *r) { + r->flags |= (RIO_FLAG_READ_ERROR | RIO_FLAG_WRITE_ERROR); +} + +/* This function allows to know if there was a read error in any past + * operation, since the rio stream was created or since the last call + * to rioClearError(). */ +static inline int rioGetReadError(rio *r) { + return (r->flags & RIO_FLAG_READ_ERROR) != 0; +} + +/* Like rioGetReadError() but for write errors. */ +static inline int rioGetWriteError(rio *r) { + return (r->flags & RIO_FLAG_WRITE_ERROR) != 0; +} + +static inline void rioClearErrors(rio *r) { + r->flags &= ~(RIO_FLAG_READ_ERROR|RIO_FLAG_WRITE_ERROR); +} + +void rioInitWithFile(rio *r, FILE *fp); +void rioInitWithBuffer(rio *r, sds s); +void rioInitWithConn(rio *r, connection *conn, size_t read_limit); +void rioInitWithFd(rio *r, int fd); +void rioInitWithConnset(rio *r, connection **conns, size_t n_conns); + +void rioFreeFd(rio *r); +void rioFreeConn(rio *r, sds* out_remainingBufferedData); +void rioFreeConnset(rio *r); + +size_t rioWriteBulkCount(rio *r, char prefix, long count); +size_t rioWriteBulkString(rio *r, const char *buf, size_t len); +size_t rioWriteBulkLongLong(rio *r, long long l); +size_t rioWriteBulkDouble(rio *r, double d); + +struct redisObject; +int rioWriteBulkObject(rio *r, struct redisObject *obj); + +void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len); +void rioSetAutoSync(rio *r, off_t bytes); +void rioSetReclaimCache(rio *r, int enabled); +uint8_t rioCheckType(rio *r); +#endif |
