tquic 线程疑问

[stopandjoin]

适用C API写了一个demo：
客户端中，在线程1中，起一个ev looper，tquic的主逻辑都在此线程中跑。 在线程2中，每隔一段时间通过 quic_stream_write 向多个 stream中写入数据。 发现会有如下问题: 线程2中第一次call quic_stream_write 时，返回正常， 后面每次都会返回 -6 ；
服务端中，读取数据并打印。 服务端只有 connection established 回调，并没有回调 stream create 和 readable。
我的疑问是：
1，tquic内部实现都是在一个单线程中吗？
2，quic_stream_xxx 接口不能在tquic本身所在线程的其他线程中调用吗？

[glime-ai[bot]]

对于您的问题，基于TQUIC的设计和文档描述，以下是对您的疑问的解答：

1. TQUIC内部实现都是在一个单线程中吗？

   • TQUIC的设计允许它在多线程环境中运行，但是它的核心网络IO处理逻辑通常在单个线程（事件循环线程）中执行。这是为了简化状态管理和避免并发访问导致的问题。然而，TQUIC支持在多线程环境中使用，特别是在处理计算密集型任务或者需要与UI线程等进行交互时。关键是要确保对TQUIC API的调用是线程安全的，或者通过某种机制（如任务队列）将调用序列化到TQUIC的主事件循环线程。

2. quic_stream_xxx 接口不能在tquic本身所在线程的其他线程中调用吗？

   • 一般来说，直接从非TQUIC主线程调用quic_stream_xxx接口可能会引起线程安全问题，因为TQUIC的状态管理和网络IO操作通常都假定在单个线程（即其主事件循环线程）中执行。如果需要从其他线程调用这些接口，应该采取措施确保这些调用在TQUIC的主线程中安全执行。这可以通过使用线程安全的队列将操作从其他线程发布到TQUIC的主线程，或者使用其他同步机制来实现。

关于您遇到的quic_stream_write返回-6的问题，这可能是因为从非主线程调用导致的线程安全问题，或者是因为流的状态不允许写入操作。建议检查TQUIC的日志以获取更详细的错误信息，并确保所有对TQUIC API的调用都遵循线程安全的原则。

总的来说，虽然TQUIC设计为支持多线程环境，但是直接从多个线程操作TQUIC的API需要小心处理，以避免线程安全问题。建议将所有TQUIC相关的操作都封装在一个安全的机制中，确保它们在TQUIC的主事件循环线程中执行。

[stopandjoin]

嗯，不设置结束标志就返回正常啦， 但quic中并没有发出去，估计我的looper哪里设置的有点问题，我再看看，谢谢啦。

[stopandjoin]

@iyangsj 添加了一些log，发现 quic_stream_write 只有在 readable的回调中写入，最终才会真实的写到socket中去。 quic_stream_write 如果是被其他异步事件（如ev async 事件）触发，虽然返回值显示写入了预期的长度。 但是socket 的write回调并没有调用。

如下，在client readable中通过quic_stream_write 写入，client和server成功进行了一次数据交互之后。然后开始被ev async事件触发，以下log显示后面Tquic中没有调到socket write，最终会在超时5秒后，断连。

_client--> read_callback
TEST_Client[140673127806720] || tquic::endpoint: CLIENT process 1 tickable connections
[140673127806720] client-->stream[12] HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK

[140673127806720] client-->stream[4] HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK

[140673127806720] client-->stream[8] HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK
HTTP/0.9 200 OK

[140673127806720] client-->stream[0] HTTP/0.9 200 OK
HTTP/0.9 200 OK_
---------------------------------------------------------------《 以上client 与 server进行了一次数据交互。
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 marked sendable true
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 next timeout duration Some(4.99992228s)
TEST_Client[140673127806720] || tquic::connection::stream: CLIENT-0371bb19194e3a51 no any readable stream
TEST_Client[140673127806720] || tquic::connection::stream: CLIENT-0371bb19194e3a51 no any writable stream
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 marked tickable false
TEST_Client[140673127806720] || tquic::endpoint: CLIENT process 1 sendable connections
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 sent packet OneRTT dcid=a4dffcf3f0b6b753 key_phase=false pn=22 frames=[ACK delay=13 ranges=[11..22] ecn_counts=None] sent_size=31 local=0.0.0.0:0 remote=127.0.0.1:4433
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 marked sendable false
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 next timeout duration Some(4.999879284s)
TEST_Client[140673127806720] || tquic::endpoint: CLIENT try to send remaining packets in packet queue
server--> recv would block
client--> client_on_packets_send :1
TEST_Client[140673127806720] || tquic::endpoint: CLIENT send total 1 packets out
client--> process_connections timer:4.999000
-----------------------------------------------------》以下 开始在异步回调中执行 quic_stream_write
async_cb in [140673127806720]
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 marked tickable true
quic_stream_write stream 0, ret:4096
quic_stream_write stream 4, ret:4096
quic_stream_write stream 8, ret:4096
quic_stream_write stream 12, ret:4096
quic_stream_write stream 16, ret:4096
quic_stream_write stream 20, ret:4096

再后面多次同样的异步调用，都显示返回值正常，但是没有tquic内部的log啦， 最后超时断开 ---》
server--> stream closed 12
server--> stream closed 4
server--> stream closed 20
server--> stream closed 8
server--> stream closed 0
server--> stream closed 16
server--> connection closed
TEST_Client[140673127806720] || tquic::endpoint: CLIENT process timeout event, connection index 0 trace id CLIENT-0371bb19194e3a51
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 timer Idle timeout
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 idle timeout
TEST_Client[140673127806720] || tquic::endpoint: CLIENT process 1 tickable connections
TEST_Client[140673127806720] || tquic::connection: CLIENT-0371bb19194e3a51 marked tickable false
TEST_Client[140673127806720] || tquic::endpoint: CLIENT process 0 sendable connections
TEST_Client[140673127806720] || tquic::endpoint: CLIENT send total 0 packets out
client--> process_connections timer:18446744949882880.000000

[stopandjoin]

补充一下信息：发现stream create后，2ms内就需要write， 超过就很容易发不出去了。 当quic_endpoint_timeout 返回值到达到 5 秒时client 再调用quic_stream_write虽返回正常 ，但socket write不会被触发。

[iyangsj]

把相关代码发出来看看，如果不便公开，可以单发 iyangsj@gmail.com

[stopandjoin]

@iyangsj 方便的，就是在 tquic-example-c 这个基础上，做了一点小修改：

Client 代码

// Copyright (c) 2023 The TQUIC Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <errno.h>
#include <ev.h>
#include <fcntl.h>
#include <inttypes.h>
#include <netdb.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

#include "openssl/ssl.h"
#include "tquic.h"
#include <pthread.h>
#include <netinet/in.h>

#define READ_BUF_SIZE 4096
#define MAX_DATAGRAM_SIZE 1200

// A simple client that supports HTTP/0.9 over QUIC
struct simple_client {
struct quic_endpoint_t *quic_endpoint;
ev_timer timer;
ev_timer send_timer;
int sock;
struct sockaddr_storage local_addr;
socklen_t local_addr_len;
struct addrinfo *peer;
SSL_CTX *ssl_ctx;
struct quic_conn_t *conn;
struct ev_loop *loop;
};

void client_on_conn_created(void *tctx, struct quic_conn_t *conn) {
struct simple_client *client = tctx;
client->conn = conn;
}

void client_on_conn_established(void *tctx, struct quic_conn_t *conn) {
printf("client--> client_on_conn_established!\n");

//const char *data = "GET /\r\n";
int ret0 = quic_stream_new(conn,0, 1, true);
int ret1 = quic_stream_new(conn,4, 1, true);
int ret2 = quic_stream_new(conn,8, 2, true);
quic_stream_new(conn,12, 2, true);
quic_stream_new(conn,16, 0, true);
quic_stream_new(conn,20, 0, true);
printf("client--> create stream ret: %d, %d, %d!\n", ret0,ret1,ret2);
//quic_stream_write(conn, 2, (uint8_t *)data, strlen(data), true);
//quic_stream_write(conn, 4, (uint8_t *)data, strlen(data), true);
//quic_stream_write(conn, 6, (uint8_t *)data, strlen(data), true);

}

void client_on_conn_closed(void *tctx, struct quic_conn_t *conn) {
struct simple_client *client = tctx;
ev_break(client->loop, EVBREAK_ALL);
}

void client_on_stream_created(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
struct simple_client *client = tctx;
//const char *data = "GET /\n";
//uint8_t buffer[4096];
//quic_stream_write(conn, stream_id, (uint8_t *)buffer, 4096/strlen(data)/, false);
if(stream_id == 0) {
ev_timer_start(client->loop,&client->send_timer); //测试启动定时器来 quic_stream_write
}
printf("client--> conn: 0x%x, client_on_stream_created :%ld\n", conn, stream_id);
}

void client_on_stream_readable(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
static uint8_t buf[READ_BUF_SIZE];
bool fin = false;
ssize_t r = quic_stream_read(conn, stream_id, buf, READ_BUF_SIZE, &fin);
if (r < 0) {
printf( "client--> stream[%ld] read error\n", stream_id);
return;
}

printf("[%ld] client-->stream[%ld] %.*s\n", pthread_self(),stream_id, (int)r, buf);

if (fin) {
    //const char *reason = "ok";
    //quic_conn_close(conn, true, 0, (const uint8_t *)reason, strlen(reason));
}

}

void client_on_stream_writable(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
quic_stream_wantwrite(conn, stream_id, false);
printf("client--> client_on_stream_writable :%ld\n", stream_id);
}

void client_on_stream_closed(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {}

int client_on_packets_send(void *psctx, struct quic_packet_out_spec_t *pkts,
unsigned int count) {
struct simple_client *client = psctx;

unsigned int sent_count = 0;
int i, j = 0;
for (i = 0; i < count; i++) {
    struct quic_packet_out_spec_t *pkt = pkts + i;
    for (j = 0; j < (*pkt).iovlen; j++) {
        const struct iovec *iov = pkt->iov + j;
        ssize_t sent =
            sendto(client->sock, iov->iov_base, iov->iov_len, 0,
                   (struct sockaddr *)pkt->dst_addr, pkt->dst_addr_len);

        if (sent != iov->iov_len) {
            if ((errno == EWOULDBLOCK) || (errno == EAGAIN)) {
                fprintf(stderr, "send would block, already sent: %d\n",
                        sent_count);
                return sent_count;
            }
            return -1;
        }
        sent_count++;
    }
}
printf("client--> client_on_packets_send :%ld\n", sent_count);

return sent_count;

}

static char s_alpn[0x100];

static int add_alpn(const char *alpn) {
size_t alpn_len, all_len;

alpn_len = strlen(alpn);
if (alpn_len > 255) return -1;

all_len = strlen(s_alpn);
if (all_len + 1 + alpn_len + 1 > sizeof(s_alpn)) return -1;

s_alpn[all_len] = alpn_len;
memcpy(&s_alpn[all_len + 1], alpn, alpn_len);
s_alpn[all_len + 1 + alpn_len] = '\0';
return 0;

}

int client_load_ssl_ctx(struct simple_client *client) {
add_alpn("transport");//("http/0.9");
client->ssl_ctx = SSL_CTX_new(TLS_method());
if (SSL_CTX_set_default_verify_paths(client->ssl_ctx) != 1) {
fprintf(stderr, "set default verify path failed\n");
return -1;
}
if (SSL_CTX_set_alpn_protos(client->ssl_ctx, (const unsigned char *)s_alpn,
strlen(s_alpn)) != 0) {
fprintf(stderr, "set alpn failed\n");
return -1;
}

return 0;

}

const struct quic_transport_methods_t quic_transport_methods = {
.on_conn_created = client_on_conn_created,
.on_conn_established = client_on_conn_established,
.on_conn_closed = client_on_conn_closed,
.on_stream_created = client_on_stream_created,
.on_stream_readable = client_on_stream_readable,
.on_stream_writable = client_on_stream_writable,
.on_stream_closed = client_on_stream_closed,
};

const struct quic_packet_send_methods_t quic_packet_send_methods = {
.on_packets_send = client_on_packets_send,
};

static void process_connections(struct simple_client *client) {
quic_endpoint_process_connections(client->quic_endpoint);
double timeout = quic_endpoint_timeout(client->quic_endpoint) / 1e3f;
if (timeout < 0.0001) {
timeout = 0.0001;
}
client->timer.repeat = timeout;
printf("client--> process_connections timer:%f \n",timeout);
ev_timer_again(client->loop, &client->timer);
}

static void read_callback(EV_P_ ev_io *w, int revents) {
struct simple_client *client = w->data;
static uint8_t buf[READ_BUF_SIZE];

while (true) {
    struct sockaddr_storage peer_addr;
    socklen_t peer_addr_len = sizeof(peer_addr);
    memset(&peer_addr, 0, peer_addr_len);

    ssize_t read = recvfrom(client->sock, buf, sizeof(buf), 0,
                            (struct sockaddr *)&peer_addr, &peer_addr_len);
    if (read < 0) {
        if ((errno == EWOULDBLOCK) || (errno == EAGAIN)) {
            break;
        }

        fprintf(stderr, "failed to read\n");
        return;
    }

    quic_packet_info_t quic_packet_info = {
        .src = (struct sockaddr *)&peer_addr,
        .src_len = peer_addr_len,
        .dst = (struct sockaddr *)&client->local_addr,
        .dst_len = client->local_addr_len,
    };

    int r = quic_endpoint_recv(client->quic_endpoint, buf, read,
                               &quic_packet_info);
    if (r != 0) {
        fprintf(stderr, "recv failed %d\n", r);
        continue;
    }
}
printf("client--> read_callback \n");

process_connections(client);

}

static void sender_callback(EV_P_ ev_timer *w, int revents) {
struct simple_client *client = w->data;
static int stream_id = 0 , looper_num = 0;
uint8_t buffer[4096];
if(stream_id > 20) {stream_id = 0; looper_num ++;}
int ret = quic_stream_write(client->conn, stream_id, (uint8_t *)buffer, 4096/strlen(data)/, false);
printf("sender_callback con: 0x%x , stream %d, ret:%ld\n",client->conn, stream_id , ret);
stream_id += 4;
if(looper_num > 10) ev_timer_stop(client->loop, &client->send_timer);
//ev_timer_again(client->loop, &client->send_timer);
}

static void timeout_callback(EV_P_ ev_timer *w, int revents) {
struct simple_client *client = w->data;
quic_endpoint_on_timeout(client->quic_endpoint);
process_connections(client);
}

static void debug_log(const unsigned char *line,size_t data_len, void *argp) {
//printf("TEST_Client[%lu] || %.*s", pthread_self(),(int)data_len,line);
}

static int create_socket(const char *host, const char *port,
struct addrinfo **peer, struct simple_client *client) {
const struct addrinfo hints = {.ai_family = PF_UNSPEC,
.ai_socktype = SOCK_DGRAM,
.ai_protocol = IPPROTO_UDP};
if (getaddrinfo(host, port, &hints, peer) != 0) {
fprintf(stderr, "failed to resolve host\n");
return -1;
}

int sock = socket((*peer)->ai_family, SOCK_DGRAM, 0);
if (sock < 0) {
    fprintf(stderr, "failed to create socket\n");
    return -1;
}
if (fcntl(sock, F_SETFL, O_NONBLOCK) != 0) {
    fprintf(stderr, "failed to make socket non-blocking\n");
    return -1;
}

client->local_addr_len = sizeof(client->local_addr);
if (getsockname(sock, (struct sockaddr *)&client->local_addr,
                &client->local_addr_len) != 0) {
    fprintf(stderr, "failed to get local address of socket\n");
    return -1;
};
client->sock = sock;

return 0;

}

void async_cb (EV_P_ ev_async *w, int revents) {
struct simple_client *client = w->data;
printf("async_cb in [%lu] \n",pthread_self());

ssize_t ret =0 ;
char sendbuf[4096]={0};
	for(char i = 0; i < 6; i ++){
	//quic_stream_wantwrite(client->conn, i * 4, true);
    ret = quic_stream_write(client->conn, i * 4, (uint8_t *)sendbuf, 4096/*strlen(data)*/, false);
	printf("quic_stream_write con: 0x%x , stream %d, ret:%ld\n",client->conn,i * 4, ret);
	}

}
ev_async async_watcher;

void* dealTquicInThread(void* param) {
printf("enter in dealEventInThread\n");
quic_set_logger(debug_log, NULL, QUIC_LOG_LEVEL_TRACE);
struct simple_client *client = (struct simple_client *)param;
quic_config_t *config = NULL;
int ret = 0;

// Create quic config.
config = quic_config_new();
if (config == NULL) {
    fprintf(stderr, "failed to create config\n");
    ret = -1;
    return NULL;
}
quic_config_set_max_idle_timeout(config, 5000);
quic_config_set_recv_udp_payload_size(config, MAX_DATAGRAM_SIZE);

// Create and set tls config.
if (client_load_ssl_ctx(client) != 0) {
    ret = -1;
    return NULL;
}
quic_config_set_tls_config(config, client->ssl_ctx);

// Create quic endpoint
client->quic_endpoint =
    quic_endpoint_new(config, false, &quic_transport_methods, client,
                      &quic_packet_send_methods, client);
if (client->quic_endpoint == NULL) {
    fprintf(stderr, "failed to create quic endpoint\n");
    ret = -1;
    return NULL;
}

// Init event loop.
client->loop = ev_default_loop(0);
ev_init(&client->timer, timeout_callback);
client->timer.data = client;
ev_timer_init(&client->send_timer, sender_callback, 0.001, 0.001); //stream create 后开启此定时器，时间大于0.002时，很难写成功
client->send_timer.data = client;
//client->send_timer.repeat = 3;
ret = quic_endpoint_connect(
    client->quic_endpoint, (struct sockaddr *)&client->local_addr,
    client->local_addr_len, client->peer->ai_addr, client->peer->ai_addrlen,
    NULL /* client_name*/, NULL /* session */, 0 /* session_len */,
    NULL /* token */, 0 /* token_len */, config,NULL /*index*/);
if (ret < 0) {
    fprintf(stderr, "failed to connect to client: %d\n", ret);
    ret = -1;
    return NULL;
}


process_connections(client);

// Start event loop.
ev_io watcher;
ev_io_init(&watcher, read_callback, client->sock, EV_READ);
ev_io_start(client->loop, &watcher);
watcher.data = client;

ev_async_init(&async_watcher, async_cb); // 初始化异步观察者
ev_async_start(client->loop, &async_watcher); // 其他线程异步触发此任务
async_watcher.data = client;
ev_loop(client->loop, 0);

printf("exit in  dealEventInThread\n");
return NULL;

}

int main(int argc, char *argv[]) {
if (argc < 3) {
fprintf(stderr, "%s <dest_addr> <dest_port>\n", argv[0]);
return -1;
}

// Set logger.
//quic_set_logger(debug_log, NULL, QUIC_LOG_LEVEL_TRACE);

// Create client.
struct simple_client client;
client.quic_endpoint = NULL;
client.ssl_ctx = NULL;
client.conn = NULL;
client.loop = NULL;
//quic_config_t *config = NULL;
int ret = 0;

// Create socket.
const char *host = argv[1];
const char *port = argv[2];
struct addrinfo *peer = NULL;
if (create_socket(host, port, &peer, &client) != 0) {
    ret = -1;
    goto EXIT;
}
client.peer = peer;

pthread_t m_threadId = 0;
if(pthread_create((pthread_t *)&m_threadId, NULL, dealTquicInThread, (void *)&client) != 0) {
		m_threadId = (pthread_t)(-1);
		printf("pthread_create error\n");
}
printf("pthread_create thread ID: %ld\n",m_threadId);


while(true) {
	usleep(500000);
    //ev_async_send(client.loop, &async_watcher);   //触发Tquic所在线程来发送数据，
}

if(m_threadId != -1)
pthread_join(m_threadId,NULL);
printf(" leaving in main thread\n");

EXIT:
if (peer != NULL) {
freeaddrinfo(peer);
}
if (client.ssl_ctx != NULL) {
SSL_CTX_free(client.ssl_ctx);
}
if (client.sock > 0) {
close(client.sock);
}
if (client.quic_endpoint != NULL) {
quic_endpoint_free(client.quic_endpoint);
}
if (client.loop != NULL) {
ev_loop_destroy(client.loop);
}
//if (config != NULL) {
// quic_config_free(config);
//}

return ret;

}

================================================================================

Server代码：

// Copyright (c) 2023 The TQUIC Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <errno.h>
#include <ev.h>
#include <fcntl.h>
#include <inttypes.h>
#include <netdb.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

#include "openssl/pem.h"
#include "openssl/ssl.h"
#include "openssl/x509.h"
#include "tquic.h"
#include <netinet/in.h>
#include <pthread.h>

//#ifndef IPPROTO_UDP
//#define IPPROTO_UDP 17
//#endif

#define READ_BUF_SIZE 4096
#define MAX_DATAGRAM_SIZE 1200

// A simple server that supports HTTP/0.9 over QUIC
struct simple_server {
struct quic_endpoint_t *quic_endpoint;
ev_timer timer;
int sock;
struct sockaddr_storage local_addr;
socklen_t local_addr_len;
SSL_CTX *ssl_ctx;
struct ev_loop *loop;
};

void server_on_conn_created(void *tctx, struct quic_conn_t *conn) {
fprintf(stderr, "server--> new connection created\n");
}

void server_on_conn_established(void *tctx, struct quic_conn_t *conn) {
fprintf(stderr, "server--> connection established\n");
}

void server_on_conn_closed(void *tctx, struct quic_conn_t *conn) {
fprintf(stderr, "server--> connection closed\n");
}

void server_on_stream_created(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
fprintf(stderr, "sever--> new stream created %ld\n", stream_id);
}

void server_on_stream_readable(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
static uint8_t buf[READ_BUF_SIZE];
bool fin = false;
ssize_t r = quic_stream_read(conn, stream_id, buf, READ_BUF_SIZE, &fin);
if (r < 0) {
fprintf(stderr, "stream[%ld] read error\n", stream_id);
return;
}

printf("server-->stream[%ld] Got request len:%ld \n", stream_id, r);
//printf("server--> %.*s\n", (int)r, buf);

//if (fin) {
    const char *resp = "HTTP/0.9 200 OK\n";
    ssize_t num = quic_stream_write(conn, stream_id, (uint8_t *)resp, strlen(resp), false);
	printf("server-->stream[%ld] Write Done: num[%ld]\n", stream_id, num);
//}

}

void server_on_stream_writable(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
printf("server-->stream[%ld] server_on_stream_writable\n", stream_id);
quic_stream_wantwrite(conn, stream_id, false);
}

void server_on_stream_closed(void *tctx, struct quic_conn_t *conn,
uint64_t stream_id) {
fprintf(stderr, "server--> stream closed %ld\n", stream_id);
}

int server_on_packets_send(void *psctx, struct quic_packet_out_spec_t *pkts,
unsigned int count) {
struct simple_server *server = psctx;

unsigned int sent_count = 0;
int i, j = 0;
for (i = 0; i < count; i++) {
    struct quic_packet_out_spec_t *pkt = pkts + i;
    for (j = 0; j < (*pkt).iovlen; j++) {
        const struct iovec *iov = pkt->iov + j;
        ssize_t sent =
            sendto(server->sock, iov->iov_base, iov->iov_len, 0,
                   (struct sockaddr *)pkt->dst_addr, pkt->dst_addr_len);

        if (sent != iov->iov_len) {
            if ((errno == EWOULDBLOCK) || (errno == EAGAIN)) {
                fprintf(stderr, "send would block, already sent: %d\n",
                        sent_count);
                return sent_count;
            }
            return -1;
        }
        fprintf(stderr, "server--> send packet, length %ld\n", sent);
        sent_count++;
    }
}

return sent_count;

}

SSL_CTX *server_get_default_tls_config(void *ctx) {
struct simple_server *server = ctx;
return server->ssl_ctx;
}

SSL_CTX *server_select_tls_config(void *ctx, const uint8_t *server_name,
size_t server_name_len) {
struct simple_server *server = ctx;
return server->ssl_ctx;
}

static char s_alpn[0x100];

static int add_alpn(const char *alpn) {
size_t alpn_len, all_len;

alpn_len = strlen(alpn);
if (alpn_len > 255) return -1;

all_len = strlen(s_alpn);
if (all_len + 1 + alpn_len + 1 > sizeof(s_alpn)) return -1;

s_alpn[all_len] = alpn_len;
memcpy(&s_alpn[all_len + 1], alpn, alpn_len);
s_alpn[all_len + 1 + alpn_len] = '\0';
return 0;

}

static int select_alpn(SSL *ssl, const unsigned char **out,
unsigned char *outlen, const unsigned char *in,
unsigned int inlen, void *arg) {
int r = SSL_select_next_proto((unsigned char **)out, outlen, in, inlen,
(unsigned char *)s_alpn, strlen(s_alpn));
if (r == OPENSSL_NPN_NEGOTIATED) {
return SSL_TLSEXT_ERR_OK;
} else {
return SSL_TLSEXT_ERR_ALERT_FATAL;
}
}

static const char *const CERT_FILE = "cert.crt";
static const char *const KEY_FILE = "cert.key";

int server_load_ssl_ctx(struct simple_server *server) {
add_alpn("transport");
server->ssl_ctx = SSL_CTX_new(TLS_method());
if (SSL_CTX_use_certificate_chain_file(server->ssl_ctx, CERT_FILE) != 1) {
fprintf(stderr, "failed to load cert\n");
SSL_CTX_free(server->ssl_ctx);
return -1;
}
if (SSL_CTX_use_PrivateKey_file(server->ssl_ctx, KEY_FILE,
SSL_FILETYPE_PEM) != 1) {
fprintf(stderr, "failed to load key\n");
SSL_CTX_free(server->ssl_ctx);
return -1;
}
SSL_CTX_set_alpn_select_cb(server->ssl_ctx, select_alpn, NULL);

return 0;

}

const struct quic_transport_methods_t quic_transport_methods = {
.on_conn_created = server_on_conn_created,
.on_conn_established = server_on_conn_established,
.on_conn_closed = server_on_conn_closed,
.on_stream_created = server_on_stream_created,
.on_stream_readable = server_on_stream_readable,
.on_stream_writable = server_on_stream_writable,
.on_stream_closed = server_on_stream_closed,
};

const struct quic_packet_send_methods_t quic_packet_send_methods = {
.on_packets_send = server_on_packets_send,
};

const struct quic_tls_config_select_methods_t tls_config_select_method = {
.get_default = server_get_default_tls_config,
.select = server_select_tls_config,
};

static void read_callback(EV_P_ ev_io *w, int revents) {
struct simple_server *server = w->data;
static uint8_t buf[READ_BUF_SIZE];

while (true) {
    struct sockaddr_storage peer_addr;
    socklen_t peer_addr_len = sizeof(peer_addr);
    memset(&peer_addr, 0, peer_addr_len);

    ssize_t read = recvfrom(server->sock, buf, sizeof(buf), 0,
                            (struct sockaddr *)&peer_addr, &peer_addr_len);
    if (read < 0) {
        if ((errno == EWOULDBLOCK) || (errno == EAGAIN)) {
            fprintf(stderr, "server--> recv would block\n");
            break;
        }

        fprintf(stderr, "failed to read\n");
        return;
    }

    quic_packet_info_t quic_packet_info = {
        .src = (struct sockaddr *)&peer_addr,
        .src_len = peer_addr_len,
        .dst = (struct sockaddr *)&server->local_addr,
        .dst_len = server->local_addr_len,
    };

    int r = quic_endpoint_recv(server->quic_endpoint, buf, read,
                               &quic_packet_info);
    if (r != 0) {
        fprintf(stderr, "recv failed %d\n", r);
        continue;
    }
}

quic_endpoint_process_connections(server->quic_endpoint);
double timeout = quic_endpoint_timeout(server->quic_endpoint) / 1e3f;
if (timeout < 0.0001) {
    timeout = 0.0001;
}
server->timer.repeat = timeout;
ev_timer_again(loop, &server->timer);

}

static void timeout_callback(EV_P_ ev_timer *w, int revents) {
struct simple_server *server = w->data;
quic_endpoint_on_timeout(server->quic_endpoint);
quic_endpoint_process_connections(server->quic_endpoint);

double timeout = quic_endpoint_timeout(server->quic_endpoint) / 1e3f;
if (timeout < 0.0001) {
    timeout = 0.0001;
}
server->timer.repeat = timeout;
ev_timer_again(loop, &server->timer);

}

static void debug_log(const unsigned char *line, size_t data_len, void *argp) {
//fprintf(stderr, "Test_Server || %s\n", line);
}

static int create_socket(const char *host, const char *port,
struct addrinfo **local,
struct simple_server *server) {
const struct addrinfo hints = {.ai_family = PF_UNSPEC,
.ai_socktype = SOCK_DGRAM,
.ai_protocol = IPPROTO_UDP};
if (getaddrinfo(host, port, &hints, local) != 0) {
fprintf(stderr, "failed to resolve host\n");
return -1;
}

int sock = socket((*local)->ai_family, SOCK_DGRAM, 0);
if (sock < 0) {
    fprintf(stderr, "failed to create socket\n");
    return -1;
}
if (fcntl(sock, F_SETFL, O_NONBLOCK) != 0) {
    fprintf(stderr, "failed to make socket non-blocking\n");
    return -1;
}
if (bind(sock, (*local)->ai_addr, (*local)->ai_addrlen) < 0) {
    fprintf(stderr, "failed to bind socket\n");
    return -1;
}

server->local_addr_len = sizeof(server->local_addr);
if (getsockname(sock, (struct sockaddr *)&server->local_addr,
                &server->local_addr_len) != 0) {
    fprintf(stderr, "failed to get local address of socket\n");
    return -1;
};
server->sock = sock;

return 0;

}

void* dealEventInThread(void* param) {
printf("enter in dealEventInThread\n");
struct simple_server *server = (struct simple_server *)param;
server->loop = ev_default_loop(0);
ev_init(&(server->timer), timeout_callback);
server->timer.data = server;

ev_io watcher;
ev_io_init(&watcher, read_callback, server->sock, EV_READ);
ev_io_start(server->loop, &watcher);
watcher.data = server;
ev_loop(server->loop, 0);

printf("exit in  dealEventInThread\n");
return NULL;

}

int main(int argc, char *argv[]) {
if (argc < 3) {
fprintf(stderr, "%s <listen_addr> <listen_port>\n", argv[0]);
return -1;
}

// Set logger.
quic_set_logger(debug_log, NULL, QUIC_LOG_LEVEL_TRACE);

// Create simple server.
struct simple_server server;
server.quic_endpoint = NULL;
server.ssl_ctx = NULL;
server.loop = NULL;
quic_config_t *config = NULL;
int ret = 0;

// Create socket.
const char *host = argv[1];
const char *port = argv[2];
struct addrinfo *local = NULL;
if (create_socket(host, port, &local, &server) != 0) {
    ret = -1;
    goto EXIT;
}

// Create quic config.
config = quic_config_new();
if (config == NULL) {
    ret = -1;
    goto EXIT;
}
quic_config_set_max_idle_timeout(config, 5000);
quic_config_set_recv_udp_payload_size(config, MAX_DATAGRAM_SIZE);

// Create and set tls conf selector for quic config.
if (server_load_ssl_ctx(&server) != 0) {
    ret = -1;
    goto EXIT;
}
quic_config_set_tls_selector(config, &tls_config_select_method, &server);

// Create quic endpoint
server.quic_endpoint =
    quic_endpoint_new(config, true, &quic_transport_methods, &server,
                      &quic_packet_send_methods, &server);
if (server.quic_endpoint == NULL) {
    fprintf(stderr, "failed to create quic endpoint\n");
    ret = -1;
    goto EXIT;
}

pthread_t m_threadId = 0;
if(pthread_create((pthread_t *)&m_threadId, NULL, dealEventInThread, (void *)&server) != 0) {
		m_threadId = (pthread_t)(-1);
		printf("pthread_create error\n");
}

// Start event loop.
//server.loop = ev_default_loop(0);
//ev_init(&server.timer, timeout_callback);
//server.timer.data = &server;

//ev_io watcher;
//ev_io_init(&watcher, read_callback, server.sock, EV_READ);
//ev_io_start(server.loop, &watcher);
//watcher.data = &server;
//ev_loop(server.loop, 0);
printf("pthread_create thread ID: %ld\n",m_threadId);

if(m_threadId != -1)
pthread_join(m_threadId,NULL);

EXIT:
if (local != NULL) {
freeaddrinfo(local);
}
if (server.ssl_ctx != NULL) {
SSL_CTX_free(server.ssl_ctx);
}
if (server.sock > 0) {
close(server.sock);
}
if (server.quic_endpoint != NULL) {
quic_endpoint_free(server.quic_endpoint);
}
if (server.loop != NULL) {
ev_loop_destroy(server.loop);
}
if (config != NULL) {
quic_config_free(config);
}

return ret;

}

[stopandjoin]

@iyangsj 从log看， 客户端在stream readable 回调中读出数据后， 如果不立马write什么业务数据，那个计时器就设置timeout 为5秒了（而主逻辑就是由这个计时器驱动的）。 再后面就算call stream_write, 也并不会驱动 tquic执行quic_endpoint_process_connections逻辑。 所以，这里的疑问是： stream_write不会主动驱动tquic逻辑， 需要用户手动call quic_endpoint_process_connections 吗？

[stopandjoin]

验证了一下，确实是这样的。 在每个quic_stream_write 之后，手动call一下quic_endpoint_process_connections 就正常了。