1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430
|
// Copyright (c) 2019, Redis Labs
// SPDX-FileCopyrightText: 2024 Redict Contributors
// SPDX-FileCopyrightText: 2024 Salvatore Sanfilippo <antirez at gmail dot com>
//
// SPDX-License-Identifier: BSD-3-Clause
// SPDX-License-Identifier: LGPL-3.0-only
#ifndef __REDICT_CONNECTION_H
#define __REDICT_CONNECTION_H
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <sys/uio.h>
#include "ae.h"
#define CONN_INFO_LEN 32
#define CONN_ADDR_STR_LEN 128 /* Similar to INET6_ADDRSTRLEN, hoping to handle other protocols. */
struct aeEventLoop;
typedef struct connection connection;
typedef struct connListener connListener;
typedef enum {
CONN_STATE_NONE = 0,
CONN_STATE_CONNECTING,
CONN_STATE_ACCEPTING,
CONN_STATE_CONNECTED,
CONN_STATE_CLOSED,
CONN_STATE_ERROR
} ConnectionState;
#define CONN_FLAG_CLOSE_SCHEDULED (1<<0) /* Closed scheduled by a handler */
#define CONN_FLAG_WRITE_BARRIER (1<<1) /* Write barrier requested */
#define CONN_TYPE_SOCKET "tcp"
#define CONN_TYPE_UNIX "unix"
#define CONN_TYPE_TLS "tls"
#define CONN_TYPE_MAX 8 /* 8 is enough to be extendable */
typedef void (*ConnectionCallbackFunc)(struct connection *conn);
typedef struct ConnectionType {
/* connection type */
const char *(*get_type)(struct connection *conn);
/* connection type initialize & finalize & configure */
void (*init)(void); /* auto-call during register */
void (*cleanup)(void);
int (*configure)(void *priv, int reconfigure);
/* ae & accept & listen & error & address handler */
void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask);
aeFileProc *accept_handler;
int (*addr)(connection *conn, char *ip, size_t ip_len, int *port, int remote);
int (*is_local)(connection *conn);
int (*listen)(connListener *listener);
/* create/shutdown/close connection */
connection* (*conn_create)(void);
connection* (*conn_create_accepted)(int fd, void *priv);
void (*shutdown)(struct connection *conn);
void (*close)(struct connection *conn);
/* connect & accept */
int (*connect)(struct connection *conn, const char *addr, int port, const char *source_addr, ConnectionCallbackFunc connect_handler);
int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
/* IO */
int (*write)(struct connection *conn, const void *data, size_t data_len);
int (*writev)(struct connection *conn, const struct iovec *iov, int iovcnt);
int (*read)(struct connection *conn, void *buf, size_t buf_len);
int (*set_write_handler)(struct connection *conn, ConnectionCallbackFunc handler, int barrier);
int (*set_read_handler)(struct connection *conn, ConnectionCallbackFunc handler);
const char *(*get_last_error)(struct connection *conn);
ssize_t (*sync_write)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_read)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_readline)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
/* pending data */
int (*has_pending_data)(void);
int (*process_pending_data)(void);
/* TLS specified methods */
sds (*get_peer_cert)(struct connection *conn);
} ConnectionType;
struct connection {
ConnectionType *type;
ConnectionState state;
int last_errno;
int fd;
short int flags;
short int refs;
unsigned short int iovcnt;
void *private_data;
ConnectionCallbackFunc conn_handler;
ConnectionCallbackFunc write_handler;
ConnectionCallbackFunc read_handler;
};
#define CONFIG_BINDADDR_MAX 16
/* Setup a listener by a connection type */
struct connListener {
int fd[CONFIG_BINDADDR_MAX];
int count;
char **bindaddr;
int bindaddr_count;
int port;
ConnectionType *ct;
void *priv; /* used by connection type specified data */
};
/* The connection module does not deal with listening and accepting sockets,
* so we assume we have a socket when an incoming connection is created.
*
* The fd supplied should therefore be associated with an already accept()ed
* socket.
*
* connAccept() may directly call accept_handler(), or return and call it
* at a later time. This behavior is a bit awkward but aims to reduce the need
* to wait for the next event loop, if no additional handshake is required.
*
* IMPORTANT: accept_handler may decide to close the connection, calling connClose().
* To make this safe, the connection is only marked with CONN_FLAG_CLOSE_SCHEDULED
* in this case, and connAccept() returns with an error.
*
* connAccept() callers must always check the return value and on error (C_ERR)
* a connClose() must be called.
*/
static inline int connAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
return conn->type->accept(conn, accept_handler);
}
/* Establish a connection. The connect_handler will be called when the connection
* is established, or if an error has occurred.
*
* The connection handler will be responsible to set up any read/write handlers
* as needed.
*
* If C_ERR is returned, the operation failed and the connection handler shall
* not be expected.
*/
static inline int connConnect(connection *conn, const char *addr, int port, const char *src_addr,
ConnectionCallbackFunc connect_handler) {
return conn->type->connect(conn, addr, port, src_addr, connect_handler);
}
/* Blocking connect.
*
* NOTE: This is implemented in order to simplify the transition to the abstract
* connections, but should probably be refactored out of cluster.c and replication.c,
* in favor of a pure async implementation.
*/
static inline int connBlockingConnect(connection *conn, const char *addr, int port, long long timeout) {
return conn->type->blocking_connect(conn, addr, port, timeout);
}
/* Write to connection, behaves the same as write(2).
*
* Like write(2), a short write is possible. A -1 return indicates an error.
*
* The caller should NOT rely on errno. Testing for an EAGAIN-like condition, use
* connGetState() to see if the connection state is still CONN_STATE_CONNECTED.
*/
static inline int connWrite(connection *conn, const void *data, size_t data_len) {
return conn->type->write(conn, data, data_len);
}
/* Gather output data from the iovcnt buffers specified by the members of the iov
* array: iov[0], iov[1], ..., iov[iovcnt-1] and write to connection, behaves the same as writev(3).
*
* Like writev(3), a short write is possible. A -1 return indicates an error.
*
* The caller should NOT rely on errno. Testing for an EAGAIN-like condition, use
* connGetState() to see if the connection state is still CONN_STATE_CONNECTED.
*/
static inline int connWritev(connection *conn, const struct iovec *iov, int iovcnt) {
return conn->type->writev(conn, iov, iovcnt);
}
/* Read from the connection, behaves the same as read(2).
*
* Like read(2), a short read is possible. A return value of 0 will indicate the
* connection was closed, and -1 will indicate an error.
*
* The caller should NOT rely on errno. Testing for an EAGAIN-like condition, use
* connGetState() to see if the connection state is still CONN_STATE_CONNECTED.
*/
static inline int connRead(connection *conn, void *buf, size_t buf_len) {
int ret = conn->type->read(conn, buf, buf_len);
return ret;
}
/* Register a write handler, to be called when the connection is writable.
* If NULL, the existing handler is removed.
*/
static inline int connSetWriteHandler(connection *conn, ConnectionCallbackFunc func) {
return conn->type->set_write_handler(conn, func, 0);
}
/* Register a read handler, to be called when the connection is readable.
* If NULL, the existing handler is removed.
*/
static inline int connSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
return conn->type->set_read_handler(conn, func);
}
/* Set a write handler, and possibly enable a write barrier, this flag is
* cleared when write handler is changed or removed.
* With barrier enabled, we never fire the event if the read handler already
* fired in the same event loop iteration. Useful when you want to persist
* things to disk before sending replies, and want to do that in a group fashion. */
static inline int connSetWriteHandlerWithBarrier(connection *conn, ConnectionCallbackFunc func, int barrier) {
return conn->type->set_write_handler(conn, func, barrier);
}
static inline void connShutdown(connection *conn) {
conn->type->shutdown(conn);
}
static inline void connClose(connection *conn) {
conn->type->close(conn);
}
/* Returns the last error encountered by the connection, as a string. If no error,
* a NULL is returned.
*/
static inline const char *connGetLastError(connection *conn) {
return conn->type->get_last_error(conn);
}
static inline ssize_t connSyncWrite(connection *conn, char *ptr, ssize_t size, long long timeout) {
return conn->type->sync_write(conn, ptr, size, timeout);
}
static inline ssize_t connSyncRead(connection *conn, char *ptr, ssize_t size, long long timeout) {
return conn->type->sync_read(conn, ptr, size, timeout);
}
static inline ssize_t connSyncReadLine(connection *conn, char *ptr, ssize_t size, long long timeout) {
return conn->type->sync_readline(conn, ptr, size, timeout);
}
/* Return CONN_TYPE_* for the specified connection */
static inline const char *connGetType(connection *conn) {
return conn->type->get_type(conn);
}
static inline int connLastErrorRetryable(connection *conn) {
return conn->last_errno == EINTR;
}
/* Get address information of a connection.
* remote works as boolean type to get local/remote address */
static inline int connAddr(connection *conn, char *ip, size_t ip_len, int *port, int remote) {
if (conn && conn->type->addr) {
return conn->type->addr(conn, ip, ip_len, port, remote);
}
return -1;
}
/* Format an IP,port pair into something easy to parse. If IP is IPv6
* (matches for ":"), the ip is surrounded by []. IP and port are just
* separated by colons. This the standard to display addresses within Redict. */
static inline int formatAddr(char *buf, size_t buf_len, char *ip, int port) {
return snprintf(buf, buf_len, strchr(ip,':') ?
"[%s]:%d" : "%s:%d", ip, port);
}
static inline int connFormatAddr(connection *conn, char *buf, size_t buf_len, int remote)
{
char ip[CONN_ADDR_STR_LEN];
int port;
if (connAddr(conn, ip, sizeof(ip), &port, remote) < 0) {
return -1;
}
return formatAddr(buf, buf_len, ip, port);
}
static inline int connAddrPeerName(connection *conn, char *ip, size_t ip_len, int *port) {
return connAddr(conn, ip, ip_len, port, 1);
}
static inline int connAddrSockName(connection *conn, char *ip, size_t ip_len, int *port) {
return connAddr(conn, ip, ip_len, port, 0);
}
/* Test a connection is local or loopback.
* Return -1 on failure, 0 is not a local connection, 1 is a local connection */
static inline int connIsLocal(connection *conn) {
if (conn && conn->type->is_local) {
return conn->type->is_local(conn);
}
return -1;
}
static inline int connGetState(connection *conn) {
return conn->state;
}
/* Returns true if a write handler is registered */
static inline int connHasWriteHandler(connection *conn) {
return conn->write_handler != NULL;
}
/* Returns true if a read handler is registered */
static inline int connHasReadHandler(connection *conn) {
return conn->read_handler != NULL;
}
/* Associate a private data pointer with the connection */
static inline void connSetPrivateData(connection *conn, void *data) {
conn->private_data = data;
}
/* Get the associated private data pointer */
static inline void *connGetPrivateData(connection *conn) {
return conn->private_data;
}
/* Return a text that describes the connection, suitable for inclusion
* in CLIENT LIST and similar outputs.
*
* For sockets, we always return "fd=<fdnum>" to maintain compatibility.
*/
static inline const char *connGetInfo(connection *conn, char *buf, size_t buf_len) {
snprintf(buf, buf_len-1, "fd=%i", conn == NULL ? -1 : conn->fd);
return buf;
}
/* anet-style wrappers to conns */
int connBlock(connection *conn);
int connNonBlock(connection *conn);
int connEnableTcpNoDelay(connection *conn);
int connDisableTcpNoDelay(connection *conn);
int connKeepAlive(connection *conn, int interval);
int connSendTimeout(connection *conn, long long ms);
int connRecvTimeout(connection *conn, long long ms);
/* Get cert for the secure connection */
static inline sds connGetPeerCert(connection *conn) {
if (conn->type->get_peer_cert) {
return conn->type->get_peer_cert(conn);
}
return NULL;
}
/* Initialize the redict connection framework */
int connTypeInitialize(void);
/* Register a connection type into redict connection framework */
int connTypeRegister(ConnectionType *ct);
/* Lookup a connection type by type name */
ConnectionType *connectionByType(const char *typename);
/* Fast path to get TCP connection type */
ConnectionType *connectionTypeTcp(void);
/* Fast path to get TLS connection type */
ConnectionType *connectionTypeTls(void);
/* Fast path to get Unix connection type */
ConnectionType *connectionTypeUnix(void);
/* Lookup the index of a connection type by type name, return -1 if not found */
int connectionIndexByType(const char *typename);
/* Create a connection of specified type */
static inline connection *connCreate(ConnectionType *ct) {
return ct->conn_create();
}
/* Create an accepted connection of specified type.
* priv is connection type specified argument */
static inline connection *connCreateAccepted(ConnectionType *ct, int fd, void *priv) {
return ct->conn_create_accepted(fd, priv);
}
/* Configure a connection type. A typical case is to configure TLS.
* priv is connection type specified,
* reconfigure is boolean type to specify if overwrite the original config */
static inline int connTypeConfigure(ConnectionType *ct, void *priv, int reconfigure) {
return ct->configure(priv, reconfigure);
}
/* Walk all the connection types and cleanup them all if possible */
void connTypeCleanupAll(void);
/* Test all the connection type has pending data or not. */
int connTypeHasPendingData(void);
/* walk all the connection types and process pending data for each connection type */
int connTypeProcessPendingData(void);
/* Listen on an initialized listener */
static inline int connListen(connListener *listener) {
return listener->ct->listen(listener);
}
/* Get accept_handler of a connection type */
static inline aeFileProc *connAcceptHandler(ConnectionType *ct) {
if (ct)
return ct->accept_handler;
return NULL;
}
/* Get Listeners information, note that caller should free the non-empty string */
sds getListensInfoString(sds info);
int RedictRegisterConnectionTypeSocket(void);
int RedictRegisterConnectionTypeUnix(void);
int RedictRegisterConnectionTypeTLS(void);
/* Return 1 if connection is using TLS protocol, 0 if otherwise. */
static inline int connIsTLS(connection *conn) {
return conn && conn->type == connectionTypeTls();
}
#endif /* __REDICT_CONNECTION_H */
|