חיבור וזרם של מידע¶
נניח שהלקוח שולח הודעה בת מאה bytes והשרת קורא לתוך buffer בגודל מאה. האם recv יחזיר מאה? ייתכן, אבל זאת לא ההתחייבות של TCP. מבחינת היישום יש זרם מסודר של bytes, ולא רשימת הודעות שנשמרת לפי הקריאות של הצד השני.
אם הלקוח מבצע שתי קריאות send, השרת יכול לקבל את המידע בקריאה אחת, בשתיים או ביותר. גבולות packetים ברשת אינם גבולות הודעות של המוצר. לכן נצטרך framing גם כאשר אנחנו משתמשים בפרוטוקול אמין שמסדר את המידע.
נבחר הודעה פשוטה: אורך בן ארבעה bytes בסדר network byte order, ואחריו גוף באורך הזה. נקבל רק גופים באורך 1 עד 1024. ההגבלה תיבדק לפני הקצאה, ולא אחרי שכבר ניסינו ליצור vector בגודל ששלח הלקוח.
משאב עם כללי חיים משלו¶
בWinsock מתחילים באמצעות WSAStartup, יוצרים socket ומסיימים אותו באמצעות closesocket. הערך שמסמן socket לא תקין הוא INVALID_SOCKET. לא נשתמש בCloseHandle רק משום שהמשאב נראה כמו מספר שאפשר להחזיק במחלקה.
גם שגיאות נקראות לפי החוזה המתאים. פעולות socket רבות מחזירות SOCKET_ERROR, ואת הפרט נקרא עם WSAGetLastError. הקריאה לWSAStartup מחזירה את קוד השגיאה ישירות. העטיפה שלנו צריכה לשמור על ההבדלים האלה, בדיוק כפי שעשינו עם registry וservice handles.
שרת ולקוח באותו קובץ¶
נכתוב תוכנית שמקבלת מצב server או client ובחירת IPv4 או IPv6. בכל הרצה נשתמש בכתובת loopback בלבד. השרת מקבל חיבור אחד, קורא הודעה אחת ומחזיר אותה; זה מאפשר לראות את מחזור החיים בלי להסתיר אותו בתוך framework.
שמרו את הקוד בקובץ stream.cpp.
#define WIN32_LEAN_AND_MEAN
#include <winsock2.h>
#include <ws2tcpip.h>
#include <cstdint>
#include <iostream>
#include <optional>
#include <stdexcept>
#include <string>
#include <string_view>
#pragma comment(lib, "ws2_32.lib")
[[noreturn]] void fail(const char* call, int code = WSAGetLastError()) {
throw std::runtime_error(std::string(call) + ": " + std::to_string(code));
}
struct winsock_scope {
winsock_scope() {
WSADATA data{};
int code = WSAStartup(MAKEWORD(2, 2), &data);
if (code != 0) fail("startup", code);
}
winsock_scope(const winsock_scope&) = delete;
winsock_scope& operator=(const winsock_scope&) = delete;
~winsock_scope() { WSACleanup(); }
};
struct socket_owner {
SOCKET value;
explicit socket_owner(SOCKET s) : value(s) {
if (s == INVALID_SOCKET) fail("socket/accept");
}
socket_owner(const socket_owner&) = delete;
socket_owner& operator=(const socket_owner&) = delete;
~socket_owner() { closesocket(value); }
};
struct addresses {
addrinfo* value = nullptr;
addresses() = default;
addresses(const addresses&) = delete;
addresses& operator=(const addresses&) = delete;
~addresses() { if (value) freeaddrinfo(value); }
};
void timeout(SOCKET s) {
DWORD milliseconds = 5000;
for (int option : {SO_RCVTIMEO, SO_SNDTIMEO}) {
if (setsockopt(s, SOL_SOCKET, option,
reinterpret_cast<const char*>(&milliseconds), sizeof(milliseconds)) == SOCKET_ERROR)
fail("timeout");
}
}
void send_all(SOCKET s, const char* bytes, int length) {
int done = 0;
while (done < length) {
int count = send(s, bytes + done, length - done, 0);
if (count == SOCKET_ERROR) fail("send");
if (count == 0) throw std::runtime_error("send made no progress");
done += count;
}
}
bool read_exact(SOCKET s, char* bytes, int length) {
int done = 0;
while (done < length) {
int count = recv(s, bytes + done, length - done, 0);
if (count == SOCKET_ERROR) fail("recv");
if (count == 0) {
if (done == 0) return false;
throw std::runtime_error("truncated field");
}
done += count;
}
return true;
}
void send_frame(SOCKET s, std::string_view text) {
if (text.empty() || text.size() > 1024)
throw std::runtime_error("invalid frame length");
std::uint32_t length = htonl(static_cast<std::uint32_t>(text.size()));
send_all(s, reinterpret_cast<const char*>(&length), sizeof(length));
send_all(s, text.data(), static_cast<int>(text.size()));
}
std::optional<std::string> read_frame(SOCKET s) {
std::uint32_t wire_length = 0;
if (!read_exact(s, reinterpret_cast<char*>(&wire_length), sizeof(wire_length)))
return std::nullopt;
auto length = ntohl(wire_length);
if (length == 0 || length > 1024)
throw std::runtime_error("frame length rejected");
std::string body(length, '\0');
if (!read_exact(s, body.data(), static_cast<int>(body.size())))
throw std::runtime_error("missing frame body");
return body;
}
void expect_eof(SOCKET s) {
char extra = 0;
int count = recv(s, &extra, 1, 0);
if (count == SOCKET_ERROR) fail("read end");
if (count != 0) throw std::runtime_error("extra data after one-frame request");
}
void end_sending(SOCKET s) {
if (shutdown(s, SD_SEND) == SOCKET_ERROR) fail("shutdown");
}
int main(int argc, char** argv) {
try {
if (argc != 3) throw std::runtime_error("use: stream server|client 4|6");
std::string_view mode(argv[1]), family(argv[2]);
if ((mode != "server" && mode != "client") ||
(family != "4" && family != "6"))
throw std::runtime_error("invalid mode or family");
winsock_scope winsock;
addrinfo hints{};
hints.ai_family = family == "4" ? AF_INET : AF_INET6;
hints.ai_socktype = SOCK_STREAM;
hints.ai_protocol = IPPROTO_TCP;
hints.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV;
addresses resolved;
int code = getaddrinfo(family == "4" ? "127.0.0.1" : "::1",
"39091", &hints, &resolved.value);
if (code != 0) fail("address", code);
auto address = resolved.value;
socket_owner endpoint(socket(address->ai_family, SOCK_STREAM, IPPROTO_TCP));
if (mode == "server") {
int enabled = 1;
if (setsockopt(endpoint.value, SOL_SOCKET, SO_EXCLUSIVEADDRUSE,
reinterpret_cast<const char*>(&enabled), sizeof(enabled)) == SOCKET_ERROR)
fail("exclusive bind");
if (address->ai_family == AF_INET6 &&
setsockopt(endpoint.value, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<const char*>(&enabled), sizeof(enabled)) == SOCKET_ERROR)
fail("IPv6 only");
if (bind(endpoint.value, address->ai_addr,
static_cast<int>(address->ai_addrlen)) == SOCKET_ERROR) fail("bind");
if (listen(endpoint.value, 1) == SOCKET_ERROR) fail("listen");
std::cout << "waiting on loopback:39091\n";
socket_owner client(accept(endpoint.value, nullptr, nullptr));
timeout(client.value);
auto request = read_frame(client.value);
if (!request) return 0;
expect_eof(client.value);
send_frame(client.value, *request);
end_sending(client.value);
} else {
timeout(endpoint.value);
if (connect(endpoint.value, address->ai_addr,
static_cast<int>(address->ai_addrlen)) == SOCKET_ERROR) fail("connect");
send_frame(endpoint.value, "hello from the course");
end_sending(endpoint.value);
auto reply = read_frame(endpoint.value);
if (!reply) throw std::runtime_error("missing reply");
expect_eof(endpoint.value);
std::cout << *reply << '\n';
}
} catch (const std::exception& error) {
std::cerr << error.what() << '\n';
return 1;
}
}
בנו בטרמינל הפיתוח והפעילו שרת בטרמינל הראשון.
בטרמינל השני הריצו לקוח. לאחר שהשרת הסתיים, חזרו על הניסוי עם הספרה 6 בשני הצדדים.
נעקוב אחרי שני הכיוונים¶
הלקוח שולח frame ואז קורא לshutdown עם SD_SEND. הוא אומר שלא ישלח עוד bytes, אבל עדיין יכול לקבל. השרת קורא את הframe, מוודא שקיבל סוף זרם בכיוון הזה, ושולח תשובה בכיוון ההפוך. רק אחר כך השרת מסיים את צד השליחה שלו.
זהו half-close. סיום כיוון אחד אינו סגירת כל החיבור ואינו שחרור הsocket. התיעוד של shutdown מפריד בין הכיוונים; את המשאב עצמו עדיין סוגרים עם closesocket.
החוזה של הפרוטוקול שלנו מחייב הודעה אחת וסוף שליחה לפני תשובה. אם לקוח ישלח הודעה וימתין לתשובה בלי לסיים את צד השליחה, שני הצדדים ימתינו עד timeout. אין כאן תקלה בTCP: שני היישומים לא הסכימו על סדר הפעולות. בפרוטוקול של כמה הודעות באותו חיבור נגדיר סדר אחר ולא נחכה לFIN אחרי כל frame.
חלקי אינו כשל¶
בread_exact אנחנו מקדמים את המצב במספר הbytes שבאמת נקראו. תשובה חיובית קצרה ממשיכה את השדה, אפס לפני header חדש מסמן סוף נקי, ואפס באמצע שדה מסמן הודעה קטועה. החוזה של recv הוא הבסיס להבחנה הזאת.
גם send_all מתקדם לפי התוצאה בפועל. הצלחת send אינה אישור שהאפליקציה המרוחקת עיבדה את הבקשה או שמרה אותה בדיסק. היא מתארת קבלה במסלול השליחה המקומי לפי חוזה הsocket. אם אנחנו צריכים אישור עסקי, נגדיר הודעת תשובה ומזהה בקשה, כפי שעשינו בשירות המקומי. אפשר לבדוק את חוזה send.
כתובת ותחום חשיפה¶
בחרנו במפורש loopback כדי שהמעבדה לא תקשיב בכל כרטיסי הרשת. הבחירה בIPv6 בלבד נעשית לפני bind. אפשר להפעיל socket במצב dual stack בתנאים המתאימים, אבל אז צריך להבין גם כתובות IPv4 mapped ואיך מפרשים את זהות הקצה. נעמיק בכך בהמשך; התיעוד על dual stack מגדיר את אפשרות המערכת.
האפשרות SO_EXCLUSIVEADDRUSE נבחרה כדי לא לשתף את נקודת ההאזנה באופן שלא התכוונו אליו. לא נעתיק אוטומטית שימוש בSO_REUSEADDR מדוגמה של מערכת אחרת; המשמעות בWinsock צריכה להיבדק לפי הפלטפורמה והכתובת.
הגבול של הניסוי¶
השרת שלנו מטפל בלקוח אחד באופן חוסם. לקוח איטי יכול להחזיק אותו, והtimeout הוא לכל פעולת socket, לא deadline כולל לכל ההודעה. לקוח שמתקדם מעט בכל פעם עשוי להישאר הרבה יותר מחמש שניות. בהרצאה הבאה נבנה מצב לכל חיבור, מגבלות וdeadline כולל, ונחבר את הפעולות לIOCP בלי לשחרר buffers מוקדם מדי.