Brunobkr/llama.cpp_AlgMor24_github

1#include "transport.h"2#include "ggml-impl.h"3 4#ifdef _WIN325#  define WIN32_LEAN_AND_MEAN6#  ifndef NOMINMAX7#     define NOMINMAX8#  endif9#  include <windows.h>10#  include <winsock2.h>11#else12#  include <arpa/inet.h>13#  include <sys/socket.h>14#  include <sys/types.h>15#  include <netinet/in.h>16#  include <netinet/tcp.h>17#  include <netdb.h>18#  include <unistd.h>19#endif20#include <cstdlib>21#include <mutex>22#include <optional>23 24#ifdef GGML_RPC_RDMA25#  include <infiniband/verbs.h>26#  include <time.h>27#  ifndef _WIN3228#    include <poll.h>29#  endif30#endif // GGML_RPC_RDMA31 32#ifdef _WIN3233typedef SOCKET sockfd_t;34using ssize_t = __int64;35#else36typedef int sockfd_t;37#endif38 39static const char * RPC_DEBUG = std::getenv("GGML_RPC_DEBUG");40 41#define LOG_DBG(...) \42    do { if (RPC_DEBUG) GGML_LOG_DEBUG(__VA_ARGS__); } while (0)43 44#ifdef GGML_RPC_RDMA45static constexpr size_t RDMA_CHUNK    = 256 * 1024;   // 256 KiB per send/recv (fits default 8 MiB memlock)46static constexpr int    RDMA_RX_DEPTH = 24;            // pre-posted recv ring: 24 × 256 KiB = 6 MiB47static constexpr size_t RDMA_GID_SIZE = 16;            // RoCE GID / IB GID is always 16 bytes48using rdma_gid_t = std::array<uint8_t, RDMA_GID_SIZE>;49 50struct rdma_conn {51    struct ibv_context * ctx = nullptr;52    struct ibv_pd * pd  = nullptr;53    struct ibv_cq * scq = nullptr;   // send completions54    struct ibv_cq * rcq = nullptr;   // recv completions55    struct ibv_qp * qp  = nullptr;56 57    void          * tx_buf = nullptr;58    struct ibv_mr * tx_mr  = nullptr;59 60    void          * rx_buf = nullptr; // RDMA_RX_DEPTH × RDMA_CHUNK contiguous61    struct ibv_mr * rx_mr  = nullptr;62    int             rx_head = 0;63 64    uint32_t        max_inline = 0;65 66    uint8_t * rx_slot(int i) const {67        return static_cast<uint8_t *>(rx_buf) + static_cast<size_t>(i) * RDMA_CHUNK;68    }69 70    bool post_rx(int i) {71        struct ibv_sge sge = {};72        sge.addr   = (uintptr_t)rx_slot(i);73        sge.length = RDMA_CHUNK;74        sge.lkey   = rx_mr->lkey;75        struct ibv_recv_wr wr = {}, * bad = nullptr;76        wr.wr_id   = (uint64_t)i;77        wr.sg_list = &sge;78        wr.num_sge = 1;79        return ibv_post_recv(qp, &wr, &bad) == 0;80    }81 82    ~rdma_conn() {83        if (tx_mr) ibv_dereg_mr(tx_mr);84        if (rx_mr) ibv_dereg_mr(rx_mr);85        free(tx_buf);86        free(rx_buf);87        if (qp)  ibv_destroy_qp(qp);88        if (scq) ibv_destroy_cq(scq);89        if (rcq) ibv_destroy_cq(rcq);90        if (pd)  ibv_dealloc_pd(pd);91        if (ctx) ibv_close_device(ctx);92    }93};94 95// Local RDMA parameters captured during the probe phase and later consumed96// by rdma_activate() after the remote side's caps arrive via HELLO.97struct rdma_local_info {98    uint32_t qpn     = 0;99    uint32_t psn     = 0;100    uint8_t  gid[RDMA_GID_SIZE] = {};101    uint8_t  ib_port = 0;102    int      gid_idx = 0;103    enum ibv_mtu path_mtu = IBV_MTU_1024;104};105 106struct rdma_caps {107    uint32_t qpn;108    uint32_t psn;109    uint8_t  gid[RDMA_GID_SIZE];110};111 112static_assert(sizeof(rdma_caps) == RPC_CONN_CAPS_SIZE, "rdma_caps must match conn_caps size");113 114#endif // GGML_RPC_RDMA115 116struct socket_t::impl {117    impl(sockfd_t fd) : use_rdma(false), fd(fd) {}118    ~impl();119    bool send_data(const void * data, size_t size);120    bool recv_data(void * data, size_t size);121    void get_caps(uint8_t * local_caps);122    void update_caps(const uint8_t * remote_caps);123 124#ifdef GGML_RPC_RDMA125    bool tcp_peer_closed();126    std::optional<rdma_gid_t> rdma_build_target_gid();127    bool rdma_probe();128    bool rdma_activate(uint32_t remote_qpn, uint32_t remote_psn, const uint8_t * remote_gid);129    bool rdma_poll(struct ibv_cq * cq, struct ibv_wc * wc);130    bool rdma_send(const void * data, size_t size);131    bool rdma_recv(void * data, size_t size);132 133    std::unique_ptr<rdma_conn> rdma;134    rdma_local_info            rdma_local = {};135#endif // GGML_RPC_RDMA136    bool     use_rdma;137    sockfd_t fd;138};139 140socket_t::impl::~impl() {141#ifdef GGML_RPC_RDMA142    rdma.reset();143#endif // GGML_RPC_RDMA144    LOG_DBG("[%s] closing socket %d\n", __func__, this->fd);145#ifdef _WIN32146    if (fd != INVALID_SOCKET) closesocket(this->fd);147#else148    if (fd >= 0) close(this->fd);149#endif150}151 152#ifdef GGML_RPC_RDMA153 154bool socket_t::impl::tcp_peer_closed() {155    if (fd < 0) return false;156#ifndef _WIN32157    struct pollfd pfd = { fd, POLLIN | POLLRDHUP, 0 };158    int r = poll(&pfd, 1, 0);159    return r > 0 && (pfd.revents & (POLLHUP | POLLERR | POLLRDHUP));160#else161    return false;162#endif163}164 165// Build a RoCE GID-shaped 16-byte target from a TCP socket's local address.166// Used to match the socket's local IP against the kernel's GID table so that167// a single memcmp handles IPv4, IPv4-mapped IPv6, and native IPv6 uniformly:168//   AF_INET                -> ::ffff:a.b.c.d  (bytes 10-11 = 0xff, last 4 = IPv4)169//   AF_INET6 (IPv4-mapped) -> ::ffff:a.b.c.d  (already in GID shape)170//   AF_INET6 (native v6)   -> the 16-byte IPv6 address as-is171// Returns std::nullopt on unsupported family or getsockname failure.172std::optional<rdma_gid_t> socket_t::impl::rdma_build_target_gid() {173    sockaddr_storage addr = {};174    socklen_t addr_len = sizeof(addr);175    if (getsockname(fd, reinterpret_cast<sockaddr *>(&addr), &addr_len) != 0) {176        return std::nullopt;177    }178    rdma_gid_t target = {};179    if (addr.ss_family == AF_INET) {180        const auto * a = reinterpret_cast<const sockaddr_in *>(&addr);181        target[10] = 0xff;182        target[11] = 0xff;183        memcpy(&target[12], &a->sin_addr, 4);184        return target;185    }186    if (addr.ss_family == AF_INET6) {187        const auto * a = reinterpret_cast<const sockaddr_in6 *>(&addr);188        memcpy(target.data(), &a->sin6_addr, RDMA_GID_SIZE);189        return target;190    }191    return std::nullopt;192}193 194bool socket_t::impl::rdma_probe() {195    const char * dev_env = std::getenv("GGML_RDMA_DEV");196    const char * gid_env = std::getenv("GGML_RDMA_GID");197 198    auto target_gid = rdma_build_target_gid();199    if (!target_gid) {200        return false;201    }202 203    const uint8_t ib_port = 1;204    int num_devs = 0;205    ibv_device ** devs = ibv_get_device_list(&num_devs);206    if (!devs || num_devs == 0) return false;207 208    ibv_context * ibctx = nullptr;209    const char * matched_dev = nullptr;210    int gid_idx = gid_env ? atoi(gid_env) : -1;211    int gid_version = IBV_GID_TYPE_IB;  // 0 = unknown/IB212 213    for (int d = 0; d < num_devs; d++) {214        const char * dn = ibv_get_device_name(devs[d]);215        if (dev_env && strcmp(dev_env, dn) != 0) continue;216 217        ibv_context * ctx = ibv_open_device(devs[d]);218        if (!ctx) continue;219 220        ibv_port_attr pa;221        if (ibv_query_port(ctx, ib_port, &pa) != 0) { ibv_close_device(ctx); continue; }222 223        int found_gid = gid_idx;224        int found_version = IBV_GID_TYPE_IB;225        if (found_gid < 0) {226            // Find a GID on this port whose bytes equal the local TCP address227            // (IPv4 or IPv6). Prefer RoCE v2 (UDP/IP, L3-routable) over v1228            // (raw Ethernet, same-L2 only) so silent hangs on L3-routed paths229            // are avoided. ibv_query_gid_ex returns gid+type in one call.230            int v2_idx = -1;231            int v1_idx = -1;232            for (int i = 0; i < pa.gid_tbl_len; i++) {233                ibv_gid_entry entry = {};234                if (ibv_query_gid_ex(ctx, ib_port, i, &entry, 0) != 0) continue;235                if (memcmp(entry.gid.raw, target_gid->data(), RDMA_GID_SIZE) != 0) continue;236                if (entry.gid_type == IBV_GID_TYPE_ROCE_V2 && v2_idx < 0) {237                    v2_idx = i;238                } else if (entry.gid_type == IBV_GID_TYPE_ROCE_V1 && v1_idx < 0) {239                    v1_idx = i;240                }241            }242            if (v2_idx >= 0) {243                found_gid = v2_idx;244                found_version = IBV_GID_TYPE_ROCE_V2;245            } else if (v1_idx >= 0) {246                found_gid = v1_idx;247                found_version = IBV_GID_TYPE_ROCE_V1;248            }249        } else {250            // Explicit GID index from GGML_RDMA_GID — fetch its type for logging.251            ibv_gid_entry entry = {};252            if (ibv_query_gid_ex(ctx, ib_port, found_gid, &entry, 0) == 0) {253                found_version = entry.gid_type;254            }255        }256        if (found_gid >= 0) {257            ibctx = ctx;258            gid_idx = found_gid;259            gid_version = found_version;260            matched_dev = dn;261            rdma_local.path_mtu = pa.active_mtu;262            break;263        }264        ibv_close_device(ctx);265    }266    ibv_free_device_list(devs);267    if (!ibctx) return false;268 269    rdma_local.ib_port = ib_port;270    rdma_local.gid_idx = gid_idx;271 272    rdma = std::make_unique<rdma_conn>();273    rdma->ctx = ibctx;274 275    rdma->pd = ibv_alloc_pd(ibctx);276    if (!rdma->pd) return false;277 278    rdma->scq = ibv_create_cq(ibctx, 16, nullptr, nullptr, 0);279    rdma->rcq = ibv_create_cq(ibctx, RDMA_RX_DEPTH + 4, nullptr, nullptr, 0);280    if (!rdma->scq || !rdma->rcq) return false;281 282    ibv_qp_init_attr qia = {};283    qia.send_cq = rdma->scq;284    qia.recv_cq = rdma->rcq;285    qia.qp_type = IBV_QPT_RC;286    qia.cap.max_send_wr     = 4;287    qia.cap.max_recv_wr     = RDMA_RX_DEPTH + 4;288    qia.cap.max_send_sge    = 1;289    qia.cap.max_recv_sge    = 1;290    qia.cap.max_inline_data = 256;291 292    rdma->qp = ibv_create_qp(rdma->pd, &qia);293    if (!rdma->qp) return false;294    rdma->max_inline = qia.cap.max_inline_data;295 296    rdma->tx_buf = aligned_alloc(4096, RDMA_CHUNK);297    rdma->rx_buf = aligned_alloc(4096, static_cast<size_t>(RDMA_RX_DEPTH) * RDMA_CHUNK);298    if (!rdma->tx_buf || !rdma->rx_buf) return false;299 300    rdma->tx_mr = ibv_reg_mr(rdma->pd, rdma->tx_buf, RDMA_CHUNK, IBV_ACCESS_LOCAL_WRITE);301    rdma->rx_mr = ibv_reg_mr(rdma->pd, rdma->rx_buf, static_cast<size_t>(RDMA_RX_DEPTH) * RDMA_CHUNK,302                           IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);303    if (!rdma->tx_mr || !rdma->rx_mr) return false;304 305    ibv_gid local_gid;306    if (ibv_query_gid(ibctx, ib_port, gid_idx, &local_gid) != 0) return false;307 308    rdma_local.qpn = rdma->qp->qp_num;309    rdma_local.psn = rdma->qp->qp_num & 0xffffff;310    memcpy(&rdma_local.gid, &local_gid, RDMA_GID_SIZE);311 312    const char * ver_str = "";313    if (gid_version == IBV_GID_TYPE_ROCE_V2) {314        ver_str = " RoCEv2";315    } else if (gid_version == IBV_GID_TYPE_ROCE_V1) {316        ver_str = " RoCEv1";317    }318    GGML_LOG_INFO("RDMA probed: dev=%s gid=%d%s qpn=%u inline=%u\n",319                  matched_dev, gid_idx, ver_str, rdma_local.qpn, rdma->max_inline);320    return true;321}322 323// Phase 2: Given remote QPN/PSN/GID, transition QP: RESET->INIT->pre-post->RTR->RTS.324// On success, the connection is live and ready for rdma_send/rdma_recv.325bool socket_t::impl::rdma_activate(uint32_t remote_qpn, uint32_t remote_psn, const uint8_t * remote_gid) {326    // RESET -> INIT327    {328        struct ibv_qp_attr a = {};329        a.qp_state        = IBV_QPS_INIT;330        a.port_num        = rdma_local.ib_port;331        a.pkey_index      = 0;332        a.qp_access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ | IBV_ACCESS_LOCAL_WRITE;333        if (ibv_modify_qp(rdma->qp, &a,334                IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT | IBV_QP_ACCESS_FLAGS) != 0) {335            return false;336        }337    }338 339    for (int i = 0; i < RDMA_RX_DEPTH; i++) {340        if (!rdma->post_rx(i)) return false;341    }342 343    // INIT -> RTR344    {345        struct ibv_qp_attr a = {};346        a.qp_state           = IBV_QPS_RTR;347        a.path_mtu           = rdma_local.path_mtu;348        a.dest_qp_num        = remote_qpn;349        a.rq_psn             = remote_psn;350        a.max_dest_rd_atomic = 1;351        a.min_rnr_timer      = 1;352        a.ah_attr.is_global  = 1;353        memcpy(&a.ah_attr.grh.dgid, remote_gid, RDMA_GID_SIZE);354        a.ah_attr.grh.hop_limit  = 1;355        a.ah_attr.grh.sgid_index = rdma_local.gid_idx;356        a.ah_attr.dlid       = 0;357        a.ah_attr.port_num   = rdma_local.ib_port;358        if (ibv_modify_qp(rdma->qp, &a,359                IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU | IBV_QP_DEST_QPN |360                IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER) != 0) {361            return false;362        }363    }364 365    // RTR -> RTS366    {367        struct ibv_qp_attr a = {};368        a.qp_state     = IBV_QPS_RTS;369        a.timeout      = 14;370        a.retry_cnt    = 7;371        a.rnr_retry    = 7;372        a.sq_psn       = rdma_local.psn;373        a.max_rd_atomic = 1;374        if (ibv_modify_qp(rdma->qp, &a,375                IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | IBV_QP_RNR_RETRY |376                IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC) != 0) {377            return false;378        }379    }380 381    GGML_LOG_INFO("RDMA activated: qpn=%u->%u mtu=%d rx_depth=%d\n",382                  rdma_local.qpn, remote_qpn, 128 << rdma_local.path_mtu, RDMA_RX_DEPTH);383    return true;384}385 386bool socket_t::impl::rdma_poll(struct ibv_cq * cq, struct ibv_wc * wc) {387    for (uint64_t s = 0; ; s++) {388        int n = ibv_poll_cq(cq, 1, wc);389        if (n > 0) {390            if (wc->status != IBV_WC_SUCCESS) {391                GGML_LOG_ERROR("RDMA CQ wc error: status=%d (%s) vendor_err=0x%x\n",392                    wc->status, ibv_wc_status_str(wc->status), wc->vendor_err);393            }394            return wc->status == IBV_WC_SUCCESS;395        }396        if (n < 0) return false;397        if ((s & 0xFFFFF) == 0 && s > 0) {398            if (tcp_peer_closed()) {399                return false;400            }401        }402    }403}404 405bool socket_t::impl::rdma_send(const void * data, size_t size) {406    rdma_conn * c = rdma.get();407    const uint8_t * src = (const uint8_t *)data;408    size_t rem = size;409    while (rem > 0) {410        size_t chunk = std::min(rem, RDMA_CHUNK);411 412        struct ibv_sge sge = {};413        struct ibv_send_wr wr = {}, * bad = nullptr;414        wr.opcode  = IBV_WR_SEND;415        wr.sg_list = &sge;416        wr.num_sge = 1;417 418        if (chunk <= c->max_inline) {419            sge.addr   = (uintptr_t)src;420            sge.length = chunk;421            wr.send_flags = IBV_SEND_SIGNALED | IBV_SEND_INLINE;422        } else {423            memcpy(c->tx_buf, src, chunk);424            sge.addr   = (uintptr_t)c->tx_buf;425            sge.length = chunk;426            sge.lkey   = c->tx_mr->lkey;427            wr.send_flags = IBV_SEND_SIGNALED;428        }429 430        if (ibv_post_send(c->qp, &wr, &bad) != 0) return false;431        struct ibv_wc wc;432        if (!rdma_poll(c->scq, &wc)) return false;433 434        src += chunk;435        rem -= chunk;436    }437    return true;438}439 440bool socket_t::impl::rdma_recv(void * data, size_t size) {441    rdma_conn * c = rdma.get();442    uint8_t * dst = (uint8_t *)data;443    size_t rem = size;444    while (rem > 0) {445        struct ibv_wc wc;446        if (!rdma_poll(c->rcq, &wc)) return false;447 448        int slot = (int)wc.wr_id;449        size_t got = wc.byte_len;450        memcpy(dst, c->rx_slot(slot), got);451 452        if (!c->post_rx(slot)) return false;453 454        dst += got;455        rem -= got;456    }457    return true;458}459 460#endif // GGML_RPC_RDMA461 462bool socket_t::impl::send_data(const void * data, size_t size) {463#ifdef GGML_RPC_RDMA464    if (use_rdma) {465        return rdma_send(data, size);466    }467#endif468    size_t bytes_sent = 0;469    while (bytes_sent < size) {470        size_t size_to_send = std::min(size - bytes_sent, MAX_CHUNK_SIZE);471        ssize_t n = send(fd, (const char *)data + bytes_sent, size_to_send, 0);472        if (n < 0) {473            GGML_LOG_ERROR("send failed (bytes_sent=%zu, size_to_send=%zu)\n",474                           bytes_sent, size_to_send);475            return false;476        }477        bytes_sent += (size_t)n;478    }479    return true;480}481 482bool socket_t::impl::recv_data(void * data, size_t size) {483#ifdef GGML_RPC_RDMA484    if (use_rdma) {485        return rdma_recv(data, size);486    }487#endif488    size_t bytes_recv = 0;489    while (bytes_recv < size) {490        size_t size_to_recv = std::min(size - bytes_recv, MAX_CHUNK_SIZE);491        ssize_t n = recv(fd, (char *)data + bytes_recv, size_to_recv, 0);492        if (n < 0) {493            GGML_LOG_ERROR("recv failed (bytes_recv=%zu, size_to_recv=%zu)\n",494                           bytes_recv, size_to_recv);495            return false;496        }497        if (n == 0) {498            LOG_DBG("recv returned 0 (peer closed?)\n");499            return false;500        }501        bytes_recv += (size_t)n;502    }503    return true;504}505 506void socket_t::impl::get_caps(uint8_t * local_caps) {507    memset(local_caps, 0, RPC_CONN_CAPS_SIZE);508#ifdef GGML_RPC_RDMA509    rdma_local = {};510    if (rdma_probe()) {511        rdma_caps rc = {};512        rc.qpn = rdma_local.qpn;513        rc.psn = rdma_local.psn;514        memcpy(rc.gid, rdma_local.gid, RDMA_GID_SIZE);515        memcpy(local_caps, &rc, sizeof(rc));516    } else {517        rdma.reset();518    }519#endif // GGML_RPC_RDMA520}521 522void socket_t::impl::update_caps(const uint8_t * remote_caps) {523#ifdef GGML_RPC_RDMA524    if (!rdma) {525        return;526    }527    rdma_caps rc = {};528    memcpy(&rc, remote_caps, sizeof(rc));529    if (rc.qpn == 0) {530        rdma.reset();531        return;532    }533    if (rdma_activate(rc.qpn, rc.psn, rc.gid)) {534        use_rdma = true;535    } else {536        GGML_LOG_ERROR("RDMA activate failed, staying on TCP\n");537        rdma.reset();538    }539#else540    (void)remote_caps;541#endif // GGML_RPC_RDMA542}543 544 545/////////////////////////////////////////////////////////////////////////////546 547socket_t::socket_t(std::unique_ptr<impl> p) : pimpl(std::move(p)) {}548 549socket_t::~socket_t() = default;550 551bool socket_t::send_data(const void * data, size_t size) {552    return pimpl->send_data(data, size);553}554 555bool socket_t::recv_data(void * data, size_t size) {556    return pimpl->recv_data(data, size);557}558 559void socket_t::get_caps(uint8_t * local_caps) {560    return pimpl->get_caps(local_caps);561}562 563void socket_t::update_caps(const uint8_t * remote_caps) {564    return pimpl->update_caps(remote_caps);565}566 567static bool is_valid_fd(sockfd_t sockfd) {568#ifdef _WIN32569    return sockfd != INVALID_SOCKET;570#else571    return sockfd >= 0;572#endif573}574 575static bool set_no_delay(sockfd_t sockfd) {576    int flag = 1;577    // set TCP_NODELAY to disable Nagle's algorithm578    int ret = setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(int));579    return ret == 0;580}581 582static bool set_reuse_addr(sockfd_t sockfd) {583    int flag = 1;584    int ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (char *)&flag, sizeof(int));585    return ret == 0;586}587 588socket_ptr socket_t::accept() {589    auto client_socket_fd = ::accept(pimpl->fd, NULL, NULL);590    if (!is_valid_fd(client_socket_fd)) {591        return nullptr;592    }593    if (!set_no_delay(client_socket_fd)) {594        GGML_LOG_ERROR("Failed to set TCP_NODELAY\n");595        return nullptr;596    }597    return socket_ptr(new socket_t(std::make_unique<impl>(client_socket_fd)));598}599 600socket_ptr socket_t::create_server(const char * host, int port) {601    auto sockfd = socket(AF_INET, SOCK_STREAM, 0);602    if (!is_valid_fd(sockfd)) {603        return nullptr;604    }605    if (!set_reuse_addr(sockfd)) {606        GGML_LOG_ERROR("Failed to set SO_REUSEADDR\n");607        return nullptr;608    }609    if (inet_addr(host) == INADDR_NONE) {610        GGML_LOG_ERROR("Invalid host address: %s\n", host);611        return nullptr;612    }613    struct sockaddr_in serv_addr;614    serv_addr.sin_family = AF_INET;615    serv_addr.sin_addr.s_addr = inet_addr(host);616    serv_addr.sin_port = htons(port);617 618    if (bind(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {619        return nullptr;620    }621    if (listen(sockfd, 1) < 0) {622        return nullptr;623    }624    return socket_ptr(new socket_t(std::make_unique<impl>(sockfd)));625}626 627socket_ptr socket_t::connect(const char * host, int port) {628    auto sockfd = socket(AF_INET, SOCK_STREAM, 0);629    if (!is_valid_fd(sockfd)) {630        return nullptr;631    }632    if (!set_no_delay(sockfd)) {633        GGML_LOG_ERROR("Failed to set TCP_NODELAY\n");634        return nullptr;635    }636    struct sockaddr_in addr;637    addr.sin_family = AF_INET;638    addr.sin_port = htons(port);639    struct hostent * server = gethostbyname(host);640    if (server == NULL) {641        GGML_LOG_ERROR("Cannot resolve host '%s'\n", host);642        return nullptr;643    }644    memcpy(&addr.sin_addr.s_addr, server->h_addr, server->h_length);645    if (::connect(sockfd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {646        return nullptr;647    }648    return socket_ptr(new socket_t(std::make_unique<impl>(sockfd)));649}650 651#ifdef _WIN32652static std::mutex g_rpc_transport_mu;653static bool g_rpc_transport_wsa_started = false;654#endif655 656bool rpc_transport_init() {657#ifdef _WIN32658    std::lock_guard<std::mutex> lock(g_rpc_transport_mu);659    if (g_rpc_transport_wsa_started) {660        return true;661    }662    WSADATA wsaData;663    int res = WSAStartup(MAKEWORD(2, 2), &wsaData);664    if (res != 0) {665        return false;666    }667    g_rpc_transport_wsa_started = true;668    return true;669#else670    return true;671#endif672}673 674void rpc_transport_shutdown() {675#ifdef _WIN32676    std::lock_guard<std::mutex> lock(g_rpc_transport_mu);677    if (!g_rpc_transport_wsa_started) {678        return;679    }680    WSACleanup();681    g_rpc_transport_wsa_started = false;682#endif683}684