对象池

发表于2022-03-26更新于2025-11-04阅读次数

内容

关联关系
一级配置器和二级配置器的关系
类型萃取
allocator
__malloc_alloc_template静态成员的类外初始化，其中函数指针尤为麻烦

关联关系

弱关联

class Person
{
private:
    Book * pbook;
public:
    Student(){}
    ~Student(){}	//不能析构book
}

强关联

class Person
{
private:
    Book & book;
public:
    Student(Book & book) : _book(book){}
}

聚合/组合关系

class Point
{
private:
    float _x;
    float _y;
public:
    Point(float x = 0.0f, float y = 0.0f) : _x(x), _y(y){}
    Point(const Point&) = default;
    Point & operator=(const Point&) = default;
    ~Point(){}
}

new

sizeof
分配字节空间
构建返回地址

delete

析构
空间释放

二级配置器

一级配置器的问题
- 直接的malloc后的数据上下各有一个越界标记。上越界标记之外还有信息，包括申请了多少字节，还有next域、prev域以便连接到链表中进行内存管理。除此之外可能还有有效/失效(是否释放)的标记。
- 除了包装信息占空间较多之外，还有malloc时花费的时间也多。这样就造成对于比较小的对象数据在malloc时入不敷出。
- 结论就是：对于较小数据，适用于内存池来进行内存管理。
对于客户端申请较小空间（128bytes）的具体处理流程：
1. 每次配置一大块内存，并维护对应的自由链表(free-list)
2. 下次如果再有相同大小的内存需求，直接从free-list中取出若干小块。
3. 如果客户端释放小块内存，就由配置器回收到free-list中。所以配置器除了分配空间，还负责回收空间。
为了方便管理，SGI第二级配置器会主动按任何小块内存的内存需求量计算成8的倍数。（比如客户端要求30bytes，则计算为32bytes。
共维护16个free-lists，各自管理大小分别为8、16、24、32、40、48、56、64、72、80、88、96、104、112、120、128的小额区块。
free-lists的单结点结构如下：

union obj
{
  union obj * free_list_link;
  char client_data[1];	//the client sees this
}

总览

1
2
3

enum { __ALIGN = 8};
enum { __MAX_BYTES = 128};
enum { __NFREELISTS = __MAX_BYTES / __ALIGN};	// 128 / 8 = 16

template<bool threads, int inst>	//参数1：是否考虑多线程，参数2：标识
class __default_alloc_template
{
private:
    union obj	/* 只是类型定义 */
    {
        union obj * free_list_link;	//next
        char client_data[1];
    };
private:
    static obj * volatile free_list[__NFREELISTS];//volatile修饰obj*，即free_list数组中对于的数据内容。
    static char* start_free;//每次配置的一大块'内存' 的内存来源的剩余部分的起始
    static char* end_free;	//每次配置的一大块'内存' 的内存来源的剩余部分的末尾
    static size_t heap_size;//上面提到的都是内存来源，而这个内存来源都是从堆区申请的，堆区申请的总空间数要记录在案
    static size_t ROUNT_UP(size_t bytes);
    static size_t FREELIST_INDEX(size_t bytes);	//hash
    static char* chunk_alloc(size_t size, int & nobjs);//size - 对象的大小，nobjs对象的个数
    static void* refill(size_t size);	//重新填充
public:	/* 用户调用 */
    static void* allocate(size_t size);
    static void deallocate(void* p, size_t n);
    static void* reallocate(void* p, size_t old_sz, size_t new_sz);
};

成员初始化

// 对free_list[__NFREELISTS]进行初始化。
template<bool threads, int inst> //	1: __default_alloc_template是模板类，需要声明
typename __default_alloc_template<threads, inst>::obj * volatile // 2: obj是__default_alloc_template<threads, inst>类内的类，需要加上“哪个类”，并且要加上"typename"以示"obj"是个类名，而非成员。
    __default_alloc_template<threads, inst>::free_list[__NFREELISTS] = {};	// 3: free_list[__NFREELISTS]是__default_alloc_template<threads, inst>类内的成员，需要加上“哪个类”，但无需加"typename"。
// 对start_free进行初始化
template<bool threads, int inst>
char* __default_alloc_template<threads, inst>::start_free = nullptr;
// 对end_free进行初始化
template<bool threads, int inst>
char* __default_alloc_template<threads, inst>::end_free = nullptr;
// 对heap_size进行初始化
template<bool threads, int inst>
size_t __default_alloc_template<threads, inst>::heap_size = 0;

函数实现

准备工作

1	using malloc_alloc = __malloc_alloc_template<0>;

allocate

相当于单链表的头删法，返回删了的节点的地址给用户用。

static void * allocate(size_t size)
{
    if(size > (size_t)__MAX_BYTES)
    {
        return malloc_alloc::allocate(size);	//转给一级配置器。
    }
    obj * result = nullptr;
    obj * volatile * my_free_list = nullptr;	//是二级指针，要指向obj*
    my_free_list = free_list + FREELIST_INDEX(size);
    result = *my_free_list;
    if(nullptr == result)	//说明此下标处无内存块了，需要申请。
    {
        void * r = refill(ROUND_UP(size));
        return r;
    }
    *my_free_list = result->free_list_link;	//把free_list此下标的指针更新到下一个节点。相当于链表的头删法。
    return result;
}

refill

static void * refill(size_t size)
{
    int nobjs = 20;
    char * chunk = chunk_alloc(size, nobjs);
    //chunk_alloc运行之后，nobjs值可能发生改变。
    if(1 == nobjs)
    {
        //如果nobjs是1，则说明内存块正好用完，不用再处理后续内存区域的链接、移动
        return chunk;
    }
    
    //接下来，把内存块分崩离析，依次串联。
    obj * volatile * my_free_list = NULL;
    obj * result = (obj*)chunk;
    obj * current_obj = NULL, * next_obj = NULL;
    my_free_list = free_list + FREELIST_INDEX(size);
    *my_free_list = next_obj = (obj*)(chunk+size);
    for(int i = 1; ; ++i)
    {
        current_obj = next_obj;
        next_obj = (obj*)((char*)next_obj + size);
        if(i == nobjs-1)//最后一块
        {
            current_obj->free_list_link = NULL;
            break;
        }
        current_obj->free_list_link = next_obj;
    }
    return result;
}

chunk_alloc

static char* chunk_alloc(size_t size, int & nobjs)
{
    char * result = NULL;
    size_t total_bytes = size * nobjs;
    size_t bytes_left = end_free - start_free;
    if(bytes_left >= total_bytes)
    {
        result = start_free;
        start_free = start_free + total_bytes;
        return result;
    }
    else if(bytes_left >= size)
    {
        nobjs = bytes_left / size;
        total = size * nobjs;
        
        result = start_free;
        start_free = start_free + total_bytes;
        return result;
    }
    else //内存不足以分配1个对象，把剩余的插入到其他下标(剩8字节插到0, 16字节插到1, ...)，然后再另外申请一个大的
    {
        size_t bytes_to_get = 2 * total_bytes + ROUND_UP(heap_size >> 4);
        if(bytes_left > 0)
        {
            obj * volatile * my_free_list = free_list + FREELIST_INDEX(bytes_left);
            //头插
            ((obj*)start_free)->free_list_link = *my_free_list;
            *my_free_list = (obj*)start_free;
        }
        //剩余的残存空间 处理完毕，接下来申请大块内存
        start_free = (char*)malloc(bytes_to_get);
        if(NULL == start_free)	//堆区居然没有资源了！无奈之计，只能向比他之后的链表找有无剩余块。
        {
            obj * volatile * my_free_list = NULL;
            for(int i = size; i <= __MAX_BYTES; i += __ALIGN)	//注意，i从size开始，而不一定从最小的块大小8开始，因为找比你小的没有用处。得找大的。
            {
                my_free_list = free_list + FREELIST_INDEX(i);
                obj * p = NULL;
                if(NULL != p)	//找到了比他大的的剩余的了！
                {
                    *my_free_list = (*my_free_list)->free_list_link;
                    start_free = (char*)p;
                    end_free = start_free + i;
                    return chunk_alloc(size, nobjs);
                }
            }
            //如果上面的for循环没能使得其找到合适的内存块，则说明真的是“山穷水尽疑无路”了，只能转1级配置器，抛出异常或者通过handler处理。
            start_free = malloc_alloc::allocate(bytes_to_get);
        }
        end_free = start_free + bytes_to_get;
        heap_size += bytes_to_get;
        return chunk_alloc(size, nobjs);	//递归此函数，相当于重新来一遍流程
    }
}

ROUND_UP & INDEX

static size_t ROUNT_UP(size_t bytes)
{
    // 0 -> 0, 1~8 -> 8, 9~16 - > 16, 17~24 -> 24, ...
    return (bytes+ __ALIGN-1) & ~(__ALIGN - 1);
}
static size_t FREELIST_INDEX(size_t bytes)
{
    // 1~8 -> 0, 9~16 -> 1, 17~24 -> 2, ...
    return ((bytes+ __ALIGN-1) / __ALIGN) - 1;
}

测试allocate

#ifdef __USE_MALLOC
typedef __malloc_alloc_template<0> malloc_alloc;
typedef malloc_alloc alloc;
#else
typedef __default_alloc_template<0, 0> alloc;
#endif

template<class T, class Alloc>
class simple_alloc
{
public:
    static T* allocate(size_t n)	//申请n个T对象
    {
        return Alloc::allocate(sizeof(T)*n);
    }
    static T* allocate()
    {
        return Alloc::allocate(sizeof(T));
    }
    void deallocate(T *p, size_t n)
    {
        if(NULL == p)return;
        Alloc::deallocate(p, sizeof(T)*n);
    }
    void deallocate(T* p)
    {
        if(NULL == p)return;
        Alloc::deallocate(p, sizeof(T));
    }
};

引入list进行测试

#include"my_list.h"
using namespace std;
int main()
{
    xcg::my_list<char> mylist;
    
    for(int i = 0; i < 23; ++i)
    {
        mylist.push_back(i + 'a');
    }
    return 0;
}

list构建流程：

template<class _Ty, class _A = xcg::alloc>
class my_list
{
public:
    typedef _Ty							 value_type;
    typedef _Ty&						 reference;
    typedef _Ty*						 pointer;
    typedef const _Ty&					 const_reference;
    typedef const _Ty*					 const_pointer;
    
    typedef _A							 allocator_type;
    typedef xcg::simple_alloc<_Node, _A> data_allocate;		//simple_alloc中使用的是第二个参数的allocate。在这里即使用_A::allocate，默认参数是xcg::alloc，而根据开关语句，若是使用malloc_alloc则为一级配置器，若为default_alloc则用二级配置器。我们默认使用的是__default_alloc_template<0, 0>。
    
public:
    my_list() : _Head(_Buynode()), _Size(0) {}
protected:
    _Nodeptr _Buynode(_Nodeptr _parg = NULL, _Nodeptr _narg = NULL)
    {
        _Nodeptr _S = data_allocate::allocate();
        _Acc::_Prev(_S) = _parg == NULL ? _S : _parg;
        _Acc::_Next(_S) = _narg == NULL ? _S : _narg;
        return _S;
    }
private:
    _Nodeptr _Head;
    size_t _Size;
};

deallocate

static void deallocate(void * p, size_t size)	// n 表示空间大小
{
    if(size > (size_t)__MAX_BYTES)
    {
        malloc_alloc::deallocate(p, size);
        return;
    }
    //归还给配置器
    obj * q = (obj*)p;
    obj * volatile * my_free_list = free_list + FREELIST_INDEX(size);
    //头插
    q->free_list_link = *my_free_list;
    *my_free_list = q;
    return;
}

测试deallocate

iterator erase(iterator _P)
{
    _Nodeptr _S = _P++._Mynode();
    _Acc::_Next(_Acc::_Prev(_S)) = _Acc::_Next(_S);
    _Acc::_Prev(_Acc::_Next(_S)) = _Acc::_Prev(_S);
    destroy(&_Acc::_Value(_S));	//不删除节点，而是释放value域。
    _Freenode(_);
    
    return _P;
}
void _Freenode(_Nodeptr _P)
{
    data_allocate::deallocate(_P);
}
void pop_back()
{
    erase(--end());
}
void pop_front()
{
    erase(begin());
}

class Object
{
private:
    int value;
public:
    Object(int x = 0) : value(x)
    {
        cout << "create object" << this << endl;
    }
    ~Object()
    {
        cout << "destroy object" << this << endl;
    }
};
int main()
{
    xcg::my_list<Object> objlist;
    for(int i = 0; i < 10; ++i)
    {
        objlist.push_back(Object(i));
    }
    objlist.pop_back();
    return 0;
}

reallocate

static void* reallocate(void * p, size_t old_sz, size_t new_sz)
{
    if(old_sz > (size_t)__MAX_BYTES && new_sz > (size_t)__MAX_BYTES)
    {
        return malloc_alloc::reallocate(p, old_sz, new_sz);
    }
    // old_sz == 20, new_sz == 22	-->24, 24
    if(ROUND_UP(old_sz) == ROUND_UP(new_sz))
    {
        return p;
    }
    // old_sz > 128, new_sz < 128
    // old_sz < 128, new_sz < 128
    // old_sz < 128, new_sz > 128
    size_t sz = old_sz < new_sz ? old_sz : new_sz;
    void * s = allocate(new_sz);
    memmove(s, p ,sz);
    deallocate(p, old, sz);
    return s;
}

网络_TCP

发表于2022-03-25更新于2025-07-22阅读次数

四次挥手

TCP

特点：

面向连接的
1. 三次握手–在客户端connect()时
  1. 必须是三次
2. 四次挥手–在任一方close时
  1. 有时可以三次
可靠的
1. 应答确认
2. 超时重传
3. 乱序重排
4. 去重
5. 滑动窗口进行流量控制
流式服务
1. 发送和接收的次数可能不一致。
  1. 连续多次发送的数据可能会被对方一次性收到。
2. 起始末尾加标记
3. send之后recv隔开
tcp是有状态的
1. 开始closed
2. listen–connecting(三次握手中)
3. established(已完成握手)
4. FIN_WAIT_1/FIN_WAIT_2
5. TIME_WAIT
  1. 可靠地终止TCP的连接
  2. 让迟来的报文在这一段时间被识别，即收集后丢弃，以防止误传给下一个使用该端口的连接。

问题：1、TCP/IP协议详解卷一；2、UNIX网络编程卷一

三次握手，四次挥手
应答确认、超时重传机制
乱序重排、去重、滑动窗口进行流量控制
什么是粘包？怎么解决？
中间转换状态的意义？TIME_WAIT状态的意义？

TCP首部

序列号(Sequence Number)字段标识了TCP发送端到TCP接收端的数据流的一个字节，该字节代表着这个报文段对应的数据中的第一个字节。如果我们考虑在两个应用程序之间的一个方向上流动的数据流，TCP给每个字节赋予一个序列号。这个序列号是一个32位的无符号数，到达 $2^{32}-1$ 后再循环回到0。
**确认号(Acknowledgment Number)**是该发送方期待接收的下一个序列号。
- TCP可以被描述为一种“带累积正向确认的滑动窗口协议”。ACK号字段被构建用于指明在接收方已经顺序收到的最大字节(加1)。例如，如果字节1_{1024已经接收成功，而下一个报文段包含字节2049}3072，那么接收方不能使用规则的ACK号字段去发信告诉发送方它接收到了这个新报文段（先保留）。然而，现代TCP有一个选择确认(Selective ACKnowledgment, SACK)选项，可以允许接收方告诉发送方它正确地接收到了次序杂乱的数据。当与一个具有选择重发(selective repeat)能力的TCP发送方搭配时，就可以实现性能的显著改善。我们将会看到TCP是如何使用重复确认(duplicate acknowledgments)以帮助它的拥塞控制和差错控制过程的。
头部长度(Header Length)字段给出了头部的长度，此字段只占4位，计算时，以32位字(4字节)为单位，比如该字段是1111，则头部长度为 $15 * (32 / 8) = 60$ 字节。它是必需的，因为选项(Options)字段的长度是可变的。
接下来是预留位(Resv)和8个标志位，旧版本的TCP头部的标志位只用到了后6个；它们中的1个或多个可被同时启用。其中除了后6个旧版本已有的，新的2个标志位为：
1. CWR：拥塞窗口减（发送方降低它的发送速率）
2. ECE：ECN回显（发送方接收到了一个更早的拥塞通告）
窗口大小(Window Size)字段。TCP的流量控制由每个端口使用这个字段来通告一个窗口大小来完成。这个窗口大小是字节数，从ACK号指定的，也是接收方想要接收的那个字节开始。这是一个16位的字段，限制了窗口大小到65535字节，从而限制了TCP的吞吐量性能。窗口缩放(Window Scale)选项可允许对这个值进行缩放,给高速和大延迟网络提供了更大的窗口和改进性能。
TCP校验和(TCP Checksum)字段覆盖了TCP的头部、数据以及IP头部中的一些字段；是一个强制性的字段，由发端计算和存储，并由收端进行验证。TCP检验和的计算与UDP检验和的计算相似，使用一个伪首部。
紧急指针(Urgent Pointer)字段。紧急指针是一个正偏移量，和序列号字段中的值相加表示紧急数据最后一个字节的序列号。这种TCP紧急机制是发送端向另一端提供特殊标志数据的方法。
选项字段。最常见的就是“最大段大小”选项。连接的每个端点一般在它发送的第一个报文段（为了建立该连接，SYN位字段被设置的那个报文段）上指定这个选项。MSS指定该选项的发送者在相反方向上希望接收到的报文段的最大值。
TCP报文段的数据部分是可选的。当一个连接被建立和终止时，交换的报文段只包含TCP头部(可能带选项)而没有数据。
- 如果这个方向上没有数据被传输，那么一个不带任何数据的头部也会用于ACK接收到的数据（称为一个pure ACK），同时通知通信方改变窗口大小（称为一个窗口更新(window update)）。
- 当一个报文段可不带数据发送时，超时操作会因此而产生一些新情况。

标志位

URG

URG，全拼是urgent。表示紧急。此标志表示：紧急指针有效。

ACK

表示确认序号字段有效。需要注意，确认标志位和确认序号字段不一样，确认序号和序列号不一样。

既然每个传输的字节都被计数，确认序号会填写该端所期望收到的下一个序号。因此确认序号应当是上次已成功收到数据字节序号+1。
发送一个ACK与发送任何一个TCP报文段的开销是一样的，因为那个32位的ACK号字段一直都
是头部的一部分，ACK位字段也一样。
TCP/IP详解_卷1_第2版_英语原文如下：Sending an ACK costs nothing more than sending any other TCP segment because the 32-bit ACK Number field is always part of the header, as is the ACK bit field.
只有ACK标志为1时，确认序号字段才有效；一旦一个连接建立起来，确认序号字段总是被设置， ACK标志也总是被设置为1；这个ACK位字段几乎用于所有报文段，除了初始和末尾报文段。

PSH

接收方应该尽快将这个报文段交给应用层。

RST

重建连接

SYN

同步序号用来发起一个连接。

当建立一个新的连接时，SYN标志变1。序列号字段随即会包含由这个主机选择的该连接的初始序列号ISN(Initial Sequence Number)。
该主机要发送数据的第一个字节序列号为ISN+1，因为SYN标志消耗了一个序号。消耗一个序列号也意味着使用重传进行可靠传输。因此，SYN和应用程序字节（还有FIN，稍后我们将会见到）是被可靠传输的。不消耗序列号的ACKs则不是。

FIN

FIN发端完成发送任务。

如何保证数据可靠传输

那些自身不包含可靠传递数据机制的协议。它们可能会使用一种像校验和或CRC这样的数学函数来检测接收到的有差错的数据，但是它们不尝试去纠正差错。比如对于IP和UDP则根本没有实现差错纠正。对于以太网和基于其上的其他协议，协议提供一定次数的重试，如果还是不成功则放弃。

通信媒介可能会丢失或改变被传递的消息，在这种环境下的通信问题已经被研究了多年。关于这个课题的一些最重要的理论工作由香农在1948年给出。这些工作普及了术语“比特”，并成为信息理论(information theory)领域的基础，帮助我们理解了在一个有损（可能会删除或改变比特）信道里可通过的信息量的根本限制。信息理论与编码理论(coding theory)的领域密切相关，编码理论提供不同的信息编码手段，从而使得信息能在通信信道里尽量免于出错。其中有两种手段：

使用差错校正码（基本上是添加一些冗余的比特，使得即使某些比特被毁，真实的信息也可以被恢复过来）来纠正通信问题是处理差错的一种非常重要的方法。
另一种方法是简单地“尝试重新发送”，直到信息最终被接收。这种方法，称为自动重复请求(Automatic Repeat Request, ARQ)，构成了许多通信协议的基础，包括TCP在内。

ARQ和重传

如果我们考虑的不只是单个通信信道，而是几个的多跳级联，我们会发现不只会碰到前面提到的那几种差错类型（分组比特差错），而且还会有更多其他的类型。这些问题可能发生在中间路由器上，是几种在讨论IP时会遇到的问题：分组重新排序，分组复制，分组泯灭（丢失）。为在多跳通信信道（例如IP）上使用而设计的带纠错的协议必须要处理这些问题。

现在让我们来探讨能处理这些问题的协议机制。在概括性地讨论这些之后，我们会探究它们是如何被TCP在互联网上使用的。

一个直接处理分组丢失（和比特差错）的方法是重发分组直到它被正确接收。这需要种方法来判断：

接收方是否已收到分组；
接收方接收到的分组是否与之前发送方发送的一样。

接收方给发送方发信号以确定自己已经接收到一个分组，这种方法称为确认(acknowledgment, ACK)。最基本的形式是，发送方发送一个分组，然后等待一个ACK。当接收方接收到这个分组时，它发送对应的ACK。当发送方接收到这个ACK，它再发送另一个分组，这个过程就这样继续。这里会有一些有意思的问题：

发送方对一个ACK应该等待多长时间？
如果ACK丢失了怎么办？
如果分组被接收到了，但是里面有错怎么办？

第一个问题其实挺深奥的。决定去等待多长时间与发送方期待(expect)为一个ACK等待多长时间有关。现在确定这个时间可能比较困难，因此我们推迟对这个技术的讨论。

第二个问题的答案比容易：如果一个ACK丢失了，发送方不易区分ACK报文段丢失与原分组丢失的情况，所以它简单地再次发送原分组。当然，这样的话，接收方可能会接收到两个或更多的拷贝，因此它必须准备好处理这种情况（见下一段）。

至于第三个问题，我们可以借助某个编码技术来解决。使用编码来检测一个大的分组中的差错（有很大的概率）一般都很简单，仅使用比其自身小很多的一些比特即可纠正。更简单的编码一般不能纠正差错，但是能检测它们。这就是校验和与CRC会如此受欢迎的原因。然后，为了检测分组里的差错，我们使用一种校验和形式。当一个接收方接收到一个含有差错的分组时，它不发送ACK。最后，发送方重发完整到达的无差错的分组。

到目前为止即使这种简单的场景，接收方都可能接收到被传送分组的重复(duplicate)副本。这个问题要使用序列号(sequence number)来处理。基本上，在被源端发送时，每个唯一的分组都有一个新的序列号，这个序列号由分组自身一直携带着。接收方可以使用这个序列号来判断它是否已经见过这个分组，如果见过则丢弃它。

到目前为止介绍的协议是可靠的，但效率不太高。如果从发送方到接收方传递即使一个很小的分组都要用很长时间（推迟或延迟）的话（如一秒或两秒，对卫星链路来说并非不正常），考虑一下那会怎样。发送方可以注入一个分组到通信路径，然后停下来等待直到它收到ACK。这个协议因此被称为“停止和等待”。假设没有分组在传输中丢失或者这些分组没有被无可挽回地损害，该协议的吞吐量(throughput)性能（每单位时间发送的数据量）与 $M/R$ 成正比， $M$ 是分组大小， $R$ 是往返时间（RTT），即分组越小、往返时间越大则性能就越低。如果有分组丢失和损害的话，情况甚至更糟糕：“吞吐质(goodput)”（每单位时间传送的有用数据量）明显要比吞吐量要低。

对于不会损害和丢失太多分组的网络来说，低吞吐量的原因是网络经常没有处于繁忙状态。情况与使用装配流水线时不出一个完整产品就不准新的工作进入类似。流水线大部分时间是空闲的。我们进一步对比，很明显，如果我们允许同一时间有多个工作单元进人流水线，就可以做得更好。对网络通信来说也是一样的一—如果我们允许多个分组进入网络，就可以使它“更繁忙”，从而得到更高的吞吐量。
很明显，允许多个分组同时进人网络使事情变得复杂。现在发送方必须不仅要决定什么时间注入一个分组到网络中，还要考虑注入多少个。并且必须要指出在等待ACK时，怎样

TIME_WAIT

2倍的MSL时间（Maximum Segment Lifetime）。