mysql默认采用的是InnoDB，默认可以支持row-level-lock,在InnoDB中，有两种锁，读共享锁和写锁，读共享锁可以多个读操作一起进行，写锁只允许一个客户端独占，由于支持的是行锁，所以在查找的时候需要有明确的主键信息，才可以
确定是那一行，然后进行加锁，没有的话，mysql会直接锁表。
读共享锁：

1

select * from tableA where id=*** LOCK IN SHARE MODE

写锁：

1
2
3

//必须指明主主键信息，才能明确到行，如果信息不能确定到具体的行的话，只能锁表，如id <> 3,
//也会直接锁表. FOR UPDATE 仅适用于InnoDB，且必须在事务区块(BEGIN/COMMIT)中才能生效。
select * from tableA where id=*** FOR UPDATE

事务：

mysql默认采用的是自动提交模式，可以采用下面的命令来修改：

1
2

set autocommit=0  //取消自动提交
set autocommit=1  //自动提交

自动提交表示你每输入一条命令，myswl自动的执行 commit，所以每一条命令对应一个事务，如果需要多条命令作为事务的话，那么需要：

1
2
3
4
5
6

BEGIN
命令1
命令2
...
...
commit

一个事务由BEGIN开始，commit结束。事务只保证所有提交的命令的执行原子性，并不能用来作为同步，默认情况下，事务是可以一起执行的，例如事务A和B都需要读取行记录r，两个事务可以并发执行。有串行模式的事务，只支持同时只有一个
事务运行，这可以用来实现同步，但是效率很低，因为一个表同时只能有一个操作。

显示进程占用cpu以及内存的情况

1
2
3
4
5
6
7
8

-d: 表示每一次刷新的时间间隔
-c: 显示整个命令行（包括参数）而不只是显示命令名
-p: 只显示具体的进程，不显示所有的进程
在top的显示过程中，还可以输入命令进行交互：
M：以内存的占有率进行排序
P：CPU占有率进行排序
T：运行时间进行排序
d：改变刷新时间

示例：

1
2
3
4
5

top   //每隔5秒显式所有进程的资源占用情况
top -d 2  //每隔2秒显式所有进程的资源占用情况
top -c  //每隔5秒显式进程的资源占用情况，并显示进程的命令行参数(默认只有进程名)
top -p 12345 -p 6789//每隔5秒显示pid是12345和pid是6789的两个进程的资源占用情况
top -d 2 -c -p 123456 //每隔2秒显示pid是12345的进程的资源使用情况，并显式该进程启动的命令行参数

从整体上看，netstat的输出结果可以分为两个部分：

一个是Active Internet connections，称为有源TCP连接，其中”Recv-Q”和”Send-Q”指%0A的是接收队列和发送队列。这些数字一般都应该是0。如果不是则表示软件包正在队列中堆积。这种情况只能在非常少的情况见到。

另一个是Active UNIX domain sockets，称为有源Unix域套接口(和网络套接字一样，但是只能用于本机通信，性能可以提高一倍)。
Proto显示连接使用的协议,RefCnt表示连接到本套接口上的进程号,Types显示套接口的类型,State显示套接口当前的状态,Path表示连接到套接口的其它进程使用的路径名。

常见参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14

-a (all)显示所有选项，默认不显示LISTEN相关
-t (tcp)仅显示tcp相关选项
-u (udp)仅显示udp相关选项
-x Unix域socket，仅显示Unxi域相关的信息
-n 拒绝显示别名，能显示数字的全部转化成数字。
-l 仅列出有在 Listen (监听) 的服務状态

-p 显示建立相关链接的程序名
-r 显示路由信息，路由表
-e 显示扩展信息，例如uid等
-s 按各个协议进行统计
-c 每隔一个固定时间，执行该netstat命令。定时执行netcat
-i 显示网络接口列表
-ie 两个加起显示类时于ifconfig一样的详细信息

提示：LISTEN和LISTENING的状态只有用-a或者-l才能看到

netcat是网络工具中的瑞士军刀，它能通过TCP和UDP在网络中读写数据。通过与其他工具结合和重定向，你可以在脚本中以多种方式使用它。使用netcat命令所能完成的事情令人惊讶。
netcat所做的就是在两台电脑之间建立链接并返回两个数据流，在这之后所能做的事就看你的想像力了。你能建立一个服务器，传输文件，与朋友聊天，传输流媒体或者用它作为其它协议的独立客户端。

其参数有以下几个：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

-d 后台模式
-e prog 程序重定向，一旦连接，就执行-e后面的程序 [危险!!]
   我们可以通过shell绑定将Windows下的cmd和Linux的/bin/sh绑定，这样非常危险，我们整个系统对别人来说是完全透明，
   从下面的例子中我们可以看到。这也是Linux默认去掉-e参数的原因。
-z：用来扫描机器端口  nc -z -v localhost 1-1024
-w：用来表示等待多久时间（以秒为单位）
-l：表示用来作为服务器
-p：表示端口  nc localhost 12345 -p 12346(指定本地端口为12346， 连接到localhost的12345端口)
-r：随机使用端口  nc localhost 12345 -r(随机使用可用端口)，效果跟 nc localhost 12345 一样
-v：显示执行过程
-vv：更加详细的显示执行过程
-q：客户端在收到EOF时，经过多少秒才会停止，默认在收到EOF时，客户端会直接停止
-k：用在服务器端，表示当客户端连接断开时，服务器依然保持运行。默认为当客户端断开连接时，服务器端会退出.当不加该参
    数时，服务器只能有一个连接，第二个客户端会连接不上，但唯一的客户大un关闭时，服务器也就关闭了加了该参数，服务
    器可以接收多个连接，且当所有客户端退出时，服务器依然运行
-u：使用udp协议
-n：使用的是数字的ip地址，不加可以使用域名
-4：使用的是ipv4
-6：使用的是ipv6

nc可以用来作为远程文件复制：

将客户端的copy文件复制到服务器端

1
2
3
4

服务端：
nc -l 12345 > copy.txt
客户端：
nc localhost 12345 < copy.txt

创建远程连接

1
2
3
4

服务端：
$nc -l 1567 -e /bin/bash -i
客户端：
$nc 172.31.100.7 1567

向前声明：只有class可以向前声明，struct不可以。对于向前声明，只是告诉编译器这里有一个类在外部定义，具体定义在这里不知道，所以在该文件里面，只能使用向前声明的类的指针或是应用，而不能直接定义该类的对象，因为不知道该类的具体定义。（在编译期，编译器必须知道类的大小，才能在构造相应的类实例时分配具体大小的空间。注：并非只在创建实例时才必须知道类的类型的大小，编译时，也必须知道。）对于在某些文件里面，由于只需要使用到某个类的指针或是引用，而不需要定义类的对象，此时就可以使用向前声明，而不需要包含定义该类的头文件，这样可以较少一些不必要的冲突或是错误。

volatile关键字的两层语义

一旦一个共享变量（类的成员变量、类的静态成员变量）被volatile修饰之后，那么就具备了两层语义：

1）保证了不同线程对这个变量进行操作时的可见性，即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。
2）禁止进行指令重排序。
3）不保证操作原子性。

跳表的机制可以参考该链接（有时间最好自己实现一下）：
http://blog.sina.com.cn/s/blog_72995dcc01017w1t.html

一个服务器一个进程

这种模型是最原始的，服务器中只有一个进程，复制监听端口，建立链接以及处理请求，然后关闭socket。这种模型效率最低，而且不能够支持长连接，对于阻塞socket来说，效率更是底下，服务器会阻塞在一个socket上，导致其他的socket被饿死。

一个链接一个进程

服务器只负责监听端口，每当收到一个链接，就fork一个进程负责处理该链接请求。这使得服务器能够同时为多个客户服务，每一个进程一个客户，客户数量的唯一限制是操作系统对以其名义运行服务器的用户ID能够同时拥有多少子进程的限制。

预先派生子进程

相比于上一个模型，预先派生子进程可以在有链接时就可以直接的有进程为客户服务，而不需要先等到fork完成后，才能够为客户服务。派生子进程在服务器启动会的时候，会预先派生一定数量的进程，当客户链接到达时，这些进程就可以立即为客户服务。所有派生的子进程都直接监听端口，单当收到连接时，就负责为建立的链接服务。由于多个子进程同时监听一个端口，会导致惊群效应。该模式只适合在于Berkeley内核的系统，对于System V， 多个进程同时在应用同一个监听socket描述符上调用accept会出现错误。

预先派生子进程+accept上锁

该模式只是上一个模式的改进，在accept上进行加锁，使得只有一个进程可以访问accept。其他的都与上一个模式一样，只不过由于加锁，这里没有了惊群效应。（其实还是有的，只不过移到了等待锁上，有多个进程在等待加锁，如果此时锁可用，那么所有等到加锁的进程都会醒过来，再一次进行加锁竞争）

预先派生子进程+传递描述符

预先派生多个进程，但是所有子进程都关闭了监听socket，只有父进程负责监听socket，父进程与子进程通过管道进行通信。有多少个子进程就有多少个管道，父进程采用select模式，监听所有的管道文件描述符和监听socket，当有新链接时，父进程负责将新的文件描述符发给空闲的子进程。父进程维护一个子进程数组，记录子进程的空闲与否，父进程与子进程通过管道进行信息交互，子进程空闲时，写管道通知父进程，同时读管道，等待父进程写新的链接描述符给子进程。当select中的管道文件描述符可读时，父进程根据管道所属的子进程来标记子进程空闲，以方便有新连接时可以将链接分配给子进程处理。

预先派线程

该系列模型其实与预先派生子进程差不多，只不过采用的是一个客户一个线程的方式。

预先派生进程或是线程都有一个问题，就是不知道开始需要派生多少，还有，需要实时对线程或是进程数量进行监控，当数量不够的时候需要采取一定的措施。

基于事件驱动

一般地,I/O多路复用机制都依赖于一个事件多路分离器(Event Demultiplexer)。分离器对象可将来自事件源的I/O事件分离出来，并分发到对应的read/write事件处理器(Event Handler)。开发人员预先注册需要处理的事件及其事件处理器（或回调函数）；事件分离器负责将请求事件传递给事件处理器。两个与事件分离器有关的模式是Reactor和Proactor。Reactor模式采用同步IO，而Proactor采用异步IO。

在Reactor中，事件分离器负责等待文件描述符或socket为读写操作准备就绪，然后将就绪事件传递给对应的处理器，最后由处理器负责完成实际的读写工作。

而在Proactor模式中，处理器或者兼任处理器的事件分离器，只负责发起异步读写操作。IO操作本身由操作系统来完成。传递给操作系统的参数需要包括用户定义的数据缓冲区地址和数据大小，操作系统才能从中得到写出操作所需数据，或写入从socket读到的数据。事件分离器捕获IO操作完成事件，然后将事件传递给对应处理器。比如，在windows上，处理器发起一个异步IO操作，再由事件分离器等待IOCompletion事件。典型的异步模式实现，都建立在操作系统支持异步API的基础之上，我们将这种实现称为“系统级”异步或“真”异步，因为应用程序完全依赖操作系统执行真正的IO工作。

TCPServer封装了Acceptor，并且自己提供了一个newConnection函数，作为Acceptor的连接建立回调函数：

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

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{
  loop_->assertInLoopThread();
  EventLoop* ioLoop = threadPool_->getNextLoop();
  char buf[64];
  snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
  ++nextConnId_;
  string connName = name_ + buf;

  LOG_INFO << "TcpServer::newConnection [" << name_
           << "] - new connection [" << connName
           << "] from " << peerAddr.toIpPort();
  InetAddress localAddr(sockets::getLocalAddr(sockfd));
  // FIXME poll with zero timeout to double confirm the new connection
  // FIXME use make_shared if necessary
  //建立一个TCPConnection，并设置相应的回调函数
  TcpConnectionPtr conn(new TcpConnection(ioLoop,
                                          connName,
                                          sockfd,
                                          localAddr,
                                          peerAddr));

  connections_[connName] = conn;
  conn->setConnectionCallback(connectionCallback_);
  conn->setMessageCallback(messageCallback_);
  conn->setWriteCompleteCallback(writeCompleteCallback_);
  conn->setCloseCallback(
      boost::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe
  ioLoop->runInLoop(boost::bind(&TcpConnection::connectEstablished, conn));
}

跟TCPConnection一样，TCPServer对外提供了事件处理回调函数接口，并将这些回调函数注册到TCPConnection中。TCPServer还提供了一个ConnectionCallBack函数，该函数也是直接注册到TCPConnection中，用来在链接建立后执行，比如可以在设置高水位处理函数以及写完成处理函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

...
server_.setConnectionCallback(
        boost::bind(&SudokuServer::onConnection, this, _1));
...
...
void onConnection(const TcpConnectionPtr& conn)
  {
    LOG_TRACE << conn->peerAddress().toIpPort() << " -> "
        << conn->localAddress().toIpPort() << " is "
        << (conn->connected() ? "UP" : "DOWN");
    if (conn->connected())
    {
      if (tcpNoDelay_)
        conn->setTcpNoDelay(true);
      conn->setHighWaterMarkCallback(
          boost::bind(&SudokuServer::highWaterMark, this, _1, _2), 5 * 1024 * 1024);
      bool throttle = false;
      conn->setContext(throttle);
    }
  }

TCPServer通过一个map来保存所有已经建立的链接：

1
2

typedef std::map<string, TcpConnectionPtr> ConnectionMap;
ConnectionMap connections_;

channel->TCPConnection->TCPServer每一个类有着自己的任务以及需要维护的属性，层与层之间通过回调函数来进行控制，上一层通过注册回调函数来控制下一层的事件操作以及维护自己的属性，下一层负责处理自己维护的属性以及执行上一层所注册的函数，从而实现每一层的信息维护以及上一层对下一层的操作控制。

利用注册回调函数，我们可以在不需要继承任何基类的情况下，通过TCPServer实现自己的服务器。对于不同功能的服务器，只需要提供自己的相关事件处理函数并注册到TCPServer中，就可以了！

在muduo中，socket的事件分发由channel来完成，针对不同的事件，channel负责调用不同的回调函数。TCPConnection则是对channel的封装，TCPConnection用来表示一个已建立的链接，每一个TCPConnection负责一个channel。TCPConnection属于应用层的类，拥有TCP链接的属性，如对端socket信息、链接信息等。TCPConnection有着不同事件处理函数，并且对外提供三个回调函数接口，应用可以对于不同的事件注册不同的回调函数，在TCPConnection执行完自己的事件处理流程后，会调用上一层注册的相应回调函数。TCPConnection直接将自己的事件处理函数作为回调函数注册到channel中，这样对于在channel中，实际上执行的事件处理函数为TCPConnection中的事件处理函数，用户可以通过向TCPConnection注册回调函数来控制socket不同事件所应采取的操作。

TCPConnection中有着接受和输出buffer缓冲，数据的读取都需要和buffer打交道，发送数据要比读取数据难，需要注意两个地方：

1. 当没有数据可以发送的时候，需要关闭读事件，否则的话会造成busy loop。
2. 当发送的数据堆积在buffer中，client一直没有接收或是数据接收很慢造成buffer太大的时候，需要进行处理，否则的话，会造成内存浪费。

TCPConnection有两个回调函数，WriteCompleteCallBack和HighWaterMarkCallBack,当buffer没有数据发送时，可以执行WriteCompleteCallBack函数，当buffer太大时，可以执行HighWaterMarkCallBack函数（一般默认是直接关闭链接，回收内存），TCPConnection暴露这两个回调接口，提供该上一层注册。
TCPConnection的代码如下：

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

//TCPConnection维护的部分信息以及属性
EventLoop* loop_;
const string name_;
StateE state_;  // FIXME: use atomic variable
// we don't expose those classes to client.
boost::scoped_ptr<Socket> socket_;
boost::scoped_ptr<Channel> channel_;
const InetAddress localAddr_;
const InetAddress peerAddr_;
//该回调函数在链接建立时执行
ConnectionCallback connectionCallback_;
MessageCallback messageCallback_;
WriteCompleteCallback writeCompleteCallback_;
HighWaterMarkCallback highWaterMarkCallback_;
CloseCallback closeCallback_;
size_t highWaterMark_;
Buffer inputBuffer_;
Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.
boost::any context_;
bool reading_;


//建立连接
TcpConnection::TcpConnection(EventLoop* loop,
                             const string& nameArg,
                             int sockfd,
                             const InetAddress& localAddr,
                             const InetAddress& peerAddr)
  : loop_(CHECK_NOTNULL(loop)),
    name_(nameArg),
    state_(kConnecting),
    socket_(new Socket(sockfd)),
    channel_(new Channel(loop, sockfd)),
    localAddr_(localAddr),
    peerAddr_(peerAddr),
    highWaterMark_(64*1024*1024),
    reading_(true)
{
  channel_->setReadCallback(
      boost::bind(&TcpConnection::handleRead, this, _1));
  channel_->setWriteCallback(
      boost::bind(&TcpConnection::handleWrite, this));
  channel_->setCloseCallback(
      boost::bind(&TcpConnection::handleClose, this));
  channel_->setErrorCallback(
      boost::bind(&TcpConnection::handleError, this));
  LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this
            << " fd=" << sockfd;
  socket_->setKeepAlive(true);
}

void TcpConnection::connectEstablished()
{
  loop_->assertInLoopThread();
  assert(state_ == kConnecting);
  setState(kConnected);
  channel_->tie(shared_from_this());
  channel_->enableReading();

  connectionCallback_(shared_from_this());
}

//关闭连接
void TcpConnection::forceCloseInLoop()
{
  loop_->assertInLoopThread();
  if (state_ == kConnected || state_ == kDisconnecting)
  {
    // as if we received 0 byte in handleRead();
    handleClose();
  }
}

//事件处理函数

//当socket可读，但是读取数据返回0的时候，说明对端关闭了链接，此时，执行handleClose函数
void TcpConnection::handleRead(Timestamp receiveTime)
{
  loop_->assertInLoopThread();
  int savedErrno = 0;
  ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
  if (n > 0)
  {
    messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
  }
  else if (n == 0)
  {
    handleClose();
  }
  else
  {
    errno = savedErrno;
    LOG_SYSERR << "TcpConnection::handleRead";
    handleError();
  }
}

void TcpConnection::handleWrite()
{
  loop_->assertInLoopThread();
  if (channel_->isWriting())
  {
    ssize_t n = sockets::write(channel_->fd(),
                               outputBuffer_.peek(),
                               outputBuffer_.readableBytes());
    if (n > 0)
    {
      outputBuffer_.retrieve(n);
      if (outputBuffer_.readableBytes() == 0)
      {
        channel_->disableWriting();
        if (writeCompleteCallback_)
        {
          loop_->queueInLoop(boost::bind(writeCompleteCallback_, shared_from_this()));
        }
        if (state_ == kDisconnecting)
        {
          shutdownInLoop();
        }
      }
    }
    else
    {
      LOG_SYSERR << "TcpConnection::handleWrite";
      // if (state_ == kDisconnecting)
      // {
      //   shutdownInLoop();
      // }
    }
  }
  else
  {
    LOG_TRACE << "Connection fd = " << channel_->fd()
              << " is down, no more writing";
  }
}

void TcpConnection::handleClose()
{
  loop_->assertInLoopThread();
  LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
  assert(state_ == kConnected || state_ == kDisconnecting);
  // we don't close fd, leave it to dtor, so we can find leaks easily.
  setState(kDisconnected);
  channel_->disableAll();

  TcpConnectionPtr guardThis(shared_from_this());
  connectionCallback_(guardThis);
  // must be the last line
  closeCallback_(guardThis);
}

void TcpConnection::handleError()
{
  int err = sockets::getSocketError(channel_->fd());
  LOG_ERROR << "TcpConnection::handleError [" << name_
            << "] - SO_ERROR = " << err << " " << strerror_tl(err);
}

对于阻塞的socket，client和server在数据的交流时，有可能会发生死锁：

1. client发送数据给server，server发送数据给client，如果数据都很大时，那么两端一起发送数据时，会阻塞在send操作上，由于每一个连接的接受端缓冲区的大小都是有限的，当接受端的接收缓冲区满时，发送端将一直阻塞在send操作上，这样一来，两端都没有读取接收缓冲区的数据，导致一直阻塞在send操作上。从而出现死锁。
2. client发送很大的数据给server，server只是echo给client，由于数据比较大，client会阻塞在send上，此时echo给client的数据占满了client的接收缓冲区，那么server的send操作将会阻塞，此时不会读取client发送的数据，这样会导致server的接收缓冲区满了，此时client的send也会一直阻塞，从而出现死锁。
   这种原因是因为应用层没有缓冲，如果两端都先接收好完整的数据存在应用层的buffer上，那么就不会出现死锁了。两端先发送需要发送的数据的大小，接下来接收到完整数据后，就可以发送给对端了。

TCP 自连接

当在本地进行tcp连接时，有可能会出现自连接。所谓的自连接就是客户端和服务端共用一个port。出现这种情况的原因如下：

1. client发起连接，连到端口为B的服务器上。
2. 系统会选择一个port A 作为client的端口。
3. 如果port B上面没有服务器在侦听，那么client会收到一个rst恢复包，如果有的话就可以连接上。
4. 如果此时port A 等于 port B， 那么此时实际上是没有服务器在port B 上监听的，不然的话，client不会分配到port B。在这种情况下，client发送的SYN包会来到端口B上，此时由于client已经打开了端口B，系统会以为端口B上有服务器在监听，那么就直接接收SYN包，并发给client。这种情况其实就是client和server同时发起连接，client此时的状态为SYN_RECEIVE，按照TCP的三次握手，client会跟自己建立起了连接。此时就出现了自连接。

自连接是一种难以确定的情况，因为在TCP下，server和client并不是对等的，这种情况下分不清谁是谁了，所以会出现难以确定的问题。为了防止自连接，每当在本地连接后，可以根据client的端口和server的端口来判断是否是自连接。

先有TCP，再有UDP，一般情况下使用的是TCP，因为UDP没有拥塞控制，很难很好的使用网络的带宽，发多会丢包，发少有不能合理使用带宽。从编程角度来说，UDP不需要网络库，在服务端，udp只需要一个socket就可以与所有的客户端进行数据交互。而TCP需要一个连接一个socket。TCP是线程不安全的，因为TCP为字节流协议，多个线程操作一个socket会发生数据错乱。而UDP是线程安全的，只需要维护客户端的信息，由于是数据包的格式，所以多个线程写一个UDP socket的话，不会造成数据包内的数据错乱。

在TCP编程的时候，如果没有数据需要发送了，那么应该调用shutdownWrite,关闭写连接，在read返回0的时候，才调用close。如果在没有数据可发送的时候直接的调用close，会导致输入缓冲区的数据直接被清除，使得数据读取不到。
TCP是可靠的意思是协议会帮你把数据安全的发送到对端的接收端口，但是并不负责数据会被对端的程序给读取到，所以为了数据的可靠传输，我们在应用层需要进行数据的接收确认，表示数据已经被程序成功处理。

在linux下使用管道时，如果管道的一端关闭了，那么往里面写数据的话，程序会受到SIGPIPE信号，在默认的情况下，程序会直接退出。在TCP网络编程下，如果对端关闭了socket的读连接，此时往里面写数据的话，也会收到SIGPIPE 信号，所以在server端，需要忽略掉SIGPIPE信号，不然的话，收到SIGPIPE信号的时候，server会退出。

TCP网络编程三部曲：

1. 开起SO_REUSEADDR,保证程序在重启的时候可以再一次的监听端口。在TCP下，如果主动关掉连接的话，会有一个TIME_WAITE时间，此时的端口还是处于忙的状态，如果不启用SO_REUSEADDR的话，绑定该端口会发生错误。
2. 忽略SIGPIPE
3. TCP_NODELAY, 一般情况下都是默认关掉该算法，使得数据可以立刻发送到对端。

在IO多路复用的时候，服务器监听的端口需要采用非阻塞的方式，因为如果端口可读的情况下，当真正的去accept的时候，如果在可读与真正accept之间，客户端断开了连接，那么在阻塞的情况下accept会出现阻塞。其他的连接socket也应该设置为非阻塞，不然的话，服务器会阻塞在某一个socket上，使得其他的socket被饿死。

在多路复用时，如果buffer没有数据可写或是没有数据想读，那么应该取消掉相应的时间侦听，否则的话会出现busy loop。

对于发布订阅等有多个客户端连接需要发布数据的情况，需要考虑有一些很慢的客户端的情况，极端情况下，客户端有可能不收数据，导致所有的消息都存在服务器上，造成服务器内存被消耗。所有跟多个客户端打交道的场景都需要考虑这个问题，防止个别客户端拖垮整个应用。至于如何解决，这个是应用场景需要考虑的问题，不是网络库的任务。在redis中，对于每一个客户端，会有一个buffer来缓冲服务器恢复给client的消息，如果buffer满了，redis会新开一个链表，每一个节点的空间是固定的，用来存放新到来的消息，如果链表尾节点由足够空间可以存放新消息，那么直接存储消息，否则的话需要新增一个节点，用来存放新消息，并将节点连到链表的尾部。每一次在链表存储消息后，redis会检查客户端的消息缓存的总大小，如果超过配置的阈值的话，那么直接关闭客户端。

Channel

每一个Channel对象只负责一个文件描述符的IO时间分发，但她并不拥有这个文件描述符，也不会在析构的时候关闭该文件描述符。channel把不同的IO事件分发为不同的回调，例如readCallBack，writeCallback等，而回调用boost::function表示，用户无需继承channel。用户一般不直接使用，而回使用更上一层的封装，如TcpConnection。
channel的部分代码如下：

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

namespace muduo
{
namespace net
{

class EventLoop;

class Channel : boost::noncopyable
{
 public:
  typedef boost::function<void()> EventCallback;
  typedef boost::function<void(Timestamp)> ReadEventCallback;

  Channel(EventLoop* loop, int fd);
  ~Channel();

  void handleEvent(Timestamp receiveTime);
  void setReadCallback(const ReadEventCallback& cb)
  { readCallback_ = cb; }
  void setWriteCallback(const EventCallback& cb)
  { writeCallback_ = cb; }
  void setCloseCallback(const EventCallback& cb)
  { closeCallback_ = cb; }
  void setErrorCallback(const EventCallback& cb)
  { errorCallback_ = cb; }

  void enableReading() { events_ |= kReadEvent; update(); }
  void disableReading() { events_ &= ~kReadEvent; update(); }
  void enableWriting() { events_ |= kWriteEvent; update(); }
  void disableWriting() { events_ &= ~kWriteEvent; update(); }
  void disableAll() { events_ = kNoneEvent; update(); }
  bool isWriting() const { return events_ & kWriteEvent; }
  bool isReading() const { return events_ & kReadEvent; }

 private:

  EventLoop* loop_;
  const int  fd_;
  int        events_;
  int        revents_; // it's the received event types of epoll or poll
  int        index_; // used by Poller.

  ReadEventCallback readCallback_;
  EventCallback writeCallback_;
  EventCallback closeCallback_;
  EventCallback errorCallback_;
};

}
}
//revents_由poller来设置，根据触发的事件类型来执行相应的回调
void Channel::handleEventWithGuard(Timestamp receiveTime)
{
  eventHandling_ = true;
  LOG_TRACE << reventsToString();
  if ((revents_ & POLLHUP) && !(revents_ & POLLIN))
  {
    if (logHup_)
    {
      LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLHUP";
    }
    if (closeCallback_) closeCallback_();
  }

  if (revents_ & POLLNVAL)
  {
    LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLNVAL";
  }

  if (revents_ & (POLLERR | POLLNVAL))
  {
    if (errorCallback_) errorCallback_();
  }
  if (revents_ & (POLLIN | POLLPRI | POLLRDHUP))
  {
    if (readCallback_) readCallback_(receiveTime);
  }
  if (revents_ & POLLOUT)
  {
    if (writeCallback_) writeCallback_();
  }
  eventHandling_ = false;
}

Poller

Poller calss为IO multiplexing的封装。Poller并不拥有channel，channel在析构前必须自己unregister(EventLoop::removeChannel()),避免空悬指针。
Poller使用两个数据结构来记录需要监听的channel：

1
2
3
4

typedef std::map<int, Channel*> ChannelMap;
typedef std::vector<struct pollfd> PollFdList;
ChannelMap channels_;
PollFdList pollfds_;

pollfds_用来记录需要监听的文件描述符以及对应监听的事件，每一个channel在这里面都有一个元素与之对应，元素的下标就是channel成员变量中的index_,poller可以跟根据每一个channel中的index来对pollfds_进行更新(removeChannel, updateChannel等操作)。

channels_用来记录每一个文件描述符对应的channel指针，这里的key为channel的文件描述符，不可以为channel的index_,因为index_会改变。当Poller获得文件描述符的监听事件后，就根据channels_来获得有时间发生的文件描述符的channel，并设置对应channel的revents_(实际发生的事件),接着将每一个有事件的channel放进一个vector里面，最后将vector返回给EventLoop，作为activateChannels。

之所以需要返回activateChannels，而不是在遍历发生事件的文件描述符的同时执行对应的channel中的handleEvent函数，是因为handleEvent函数有可能会对pollfds_进行修改，如删除channel，使得在迭代期间pollfds_大小发生改变，这是一件很危险的事情。另一个原因就是简化Poller的职责，Poller只负责监听，不负责事件的分发与处理。

如果某一个channel暂时不关心任何事件，那么可以吧pollfd.fd设置为负数，这样poll会忽略此文件描述符。不能将pollfd.events设置为0，因为无法屏蔽POLLERR事件。muduo的改进做法是吧pollfd.fd设为channel->fd()的相反数减一（文件描述符从0开始，减一是为了兼容0，因为0的相反数还是0）。

但需要删除channel时，由于pollfds_使用的是vector来存储channel的，可以根据channle本身的index_来确定该channel在vector中所在的位置。如果直接删除该位置的话，那么后面的channel都需要上移一个位置，导致后面的channel都需要更新自己的index_,这样效率会很低。这里有一个小技巧，就是直接将vector最后的元素A与该位置进行呼唤，只需要修改A的index_就好，其他的channel不受影响。

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

void PollPoller::removeChannel(Channel* channel)
{
  Poller::assertInLoopThread();
  LOG_TRACE << "fd = " << channel->fd();
  assert(channels_.find(channel->fd()) != channels_.end());
  assert(channels_[channel->fd()] == channel);
  assert(channel->isNoneEvent());
  int idx = channel->index();
  assert(0 <= idx && idx < static_cast<int>(pollfds_.size()));
  const struct pollfd& pfd = pollfds_[idx]; (void)pfd;
  assert(pfd.fd == -channel->fd()-1 && pfd.events == channel->events());
  size_t n = channels_.erase(channel->fd());
  assert(n == 1); (void)n;
  if (implicit_cast<size_t>(idx) == pollfds_.size()-1)
  {
    pollfds_.pop_back();
  }
  else
  {
    int channelAtEnd = pollfds_.back().fd;
    iter_swap(pollfds_.begin()+idx, pollfds_.end()-1);
    if (channelAtEnd < 0)
    {
      channelAtEnd = -channelAtEnd-1;
    }
    channels_[channelAtEnd]->set_index(idx);
    pollfds_.pop_back();
  }
}

muduo主要采用Reactor模式来实现C++网络库。一个EventLoop对应一个线程。服务器就是在EventLoop中来实现对网络连接的管理的，所有的socket的读写都在该loop中实现。EventLoop的执行大体步骤如下：

代码为：

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

void EventLoop::loop()
{
  assert(!looping_);
  assertInLoopThread();
  looping_ = true;
  quit_ = false;  // FIXME: what if someone calls quit() before loop() ?
  LOG_TRACE << "EventLoop " << this << " start looping";

  while (!quit_)
  {
    activeChannels_.clear();
    pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
    ++iteration_;
    if (Logger::logLevel() <= Logger::TRACE)
    {
      printActiveChannels();
    }
    // TODO sort channel by priority
    eventHandling_ = true;
    for (ChannelList::iterator it = activeChannels_.begin();
        it != activeChannels_.end(); ++it)
    {
      currentActiveChannel_ = *it;
      currentActiveChannel_->handleEvent(pollReturnTime_);
    }
    currentActiveChannel_ = NULL;
    eventHandling_ = false;
    doPendingFunctors();
  }

  LOG_TRACE << "EventLoop " << this << " stop looping";
  looping_ = false;
}

pendingFunctors的定义如下：

1
2

typedef boost::function<void()> Functor;
std::vector<Functor> pendingFunctors_;

在EventLoop中，只有pendingFunctors暴露给其他的线程，所以对于该成员的修改需要mutex来保护

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

void EventLoop::runInLoop(Functor&& cb)
{
  if (isInLoopThread())
  {
    cb();
  }
  else
  {
    queueInLoop(std::move(cb));
  }
}

void EventLoop::queueInLoop(Functor&& cb)
{
  {
  MutexLockGuard lock(mutex_);
  pendingFunctors_.push_back(std::move(cb));  // emplace_back
  }

  if (!isInLoopThread() || callingPendingFunctors_)
  {
    wakeup();
  }
}

在遍历pendingFunctors的时候，采用的是先剪切，再遍历：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

void EventLoop::doPendingFunctors()
{
  std::vector<Functor> functors;
  callingPendingFunctors_ = true;

  {
  MutexLockGuard lock(mutex_);
  functors.swap(pendingFunctors_);
  }

  for (size_t i = 0; i < functors.size(); ++i)
  {
    functors[i]();
  }
  callingPendingFunctors_ = false;
}

这样可以在遍历的时候，不会阻塞其他线程往EventLoop里面注册函数，并且，由于回调函数里面有可能也会调用queueInLoop函数，所以这样也可以避免死循环，先剪切后，遍历的是新的functors，原来的pendingfunctors为空，如果回调函数又执行queueInLoop的话，注册的是pendingFunctors，跟新的Functors无关。所以可以安全的遍历新的Functors。

由于EventLoop有可能会一直阻塞在poll中，此时如果有回调函数需要EventLoop马上执行的话，那么需要立马唤醒EventLoop的阻塞，为了实现该功能，EventLoop使用了Linux现有的eventfd。

eventfd：实现了线程之间事件通知的方式，eventfd的缓冲区大小是sizeof(uint64_t)；向其write可以递增这个计数器，read操作可以读取，并进行清零；eventfd也可以放到监听队列中，当计数器不是0时，有可读事件发生，可以进行读取。

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

EventLoop::EventLoop()
  : looping_(false),
    quit_(false),
    eventHandling_(false),
    callingPendingFunctors_(false),
    iteration_(0),
    threadId_(CurrentThread::tid()),
    poller_(Poller::newDefaultPoller(this)),
    timerQueue_(new TimerQueue(this)),
    wakeupFd_(createEventfd()),
    wakeupChannel_(new Channel(this, wakeupFd_)),
    currentActiveChannel_(NULL)
{
  LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
  if (t_loopInThisThread)
  {
    LOG_FATAL << "Another EventLoop " << t_loopInThisThread
              << " exists in this thread " << threadId_;
  }
  else
  {
    t_loopInThisThread = this;
  }
  wakeupChannel_->setReadCallback(
      boost::bind(&EventLoop::handleRead, this));
  // we are always reading the wakeupfd
  wakeupChannel_->enableReading();
}

int createEventfd()
{
  int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
  if (evtfd < 0)
  {
    LOG_SYSERR << "Failed in eventfd";
    abort();
  }
  return evtfd;
}

void EventLoop::wakeup()
{
  uint64_t one = 1;
  ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
  if (n != sizeof one)
  {
    LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
  }
}

void EventLoop::handleRead()
{
  uint64_t one = 1;
  ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
  if (n != sizeof one)
  {
    LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
  }
}

EventLoop提供一个接口来直接退出EventLoop循环，该接口为public，允许外面的对象访问：

1
2
3
4
5
6
7
8
9
10
11

void EventLoop::quit()
{
  quit_ = true;
  // There is a chance that loop() just executes while(!quit_) and exits,
  // then EventLoop destructs, then we are accessing an invalid object.
  // Can be fixed using mutex_ in both places.
  if (!isInLoopThread())
  {
    wakeup();
  }
}

muduo在应用层上加了一个缓存机制，用户发送接受数据只需要与buffer打交道就好，底层的网络通信由buffer来实现。

为什么non-blocking网络编程中应用层buffer是必须的

non-blocking IO的核心思想就是避免阻塞在read()或write()或其他的IO系统调用上，这样可以最大限度的复用thread-of-control，让一个线程能服务于多个sockets链接。IO线程只能阻塞在IO multiplexing函数上，如select/poll/epoll_wait.这样一来，应用层的缓存是必须的。每一个TCP socket都要有stateful的input buffer和output buffer。

TCPConnection必须要有output buffer

考虑一个常见的场景：程序想要通过TCP联系发送100KB的数据，但是write()调用中，操作系统只接受80KB的，此时如果直接让应用程序发送的话，那么剩下的20KB数据就发不出去，应用程序可以等到发送缓冲区有空间的时候继续发送，但是不知道什么时候有空间，此时将会阻塞，如果不等待的话，应用程序需要记录剩下的20KB，等到下一次发，然后继续循环上面的操作。但是应用程序只是负责生产数据，对于数据什么时候发送，以怎样的大小发送，应用程序并不关心，这些事情应该都由网络库来实现，所以需要有output buffer。

TcpConnection必须要有input buffer

TCP是一个无边界的字节流协议，接受方必须要处理“收到的数据尚不构成一条完整的消息”和“一次性收到两条消息的数据”等情况。网络库在处理socket可读事件的时候，必须一次性吧socket里的数据读完，否则会反复触发POLLIN事件，造成busy-loop。那么网络库就需要应对数据不完整的情况，需要将收到的数据放在input buffer中，等构成一条完整的消息后，在通知应用程序。也就是应用层的消息分包

Buffer的类图

buffer内部数据存储结构为vector,支持类似于队列的先进先出操作。具体的类图如下：

Buffer::readFD()

在非阻塞网络编程中，对于缓冲去大小的设定需要根据不同应用的场景来决定。为了减少系统调用，一次性读的数据越多约好，但是需要一个比较大的缓冲区。另一方面，为了减少系统的内存占用，应该将缓冲区设的小一点比较好，不然的话，如果一个连接的缓冲区大小为1M，那么1000个连接就需要21G的内存，但是在大多数情况下，buffer的使用率很低，此时大的buffer会造成内存的浪费。

在muduo里面，readFD函数的实现为：在栈上准备一个65536字节的extrabuf，然后利用readv()来读取数据，iovec有两块，一个指向input buffer的writable字节，一个指向extrabuf。如果读入的数据很多，那么可以放在extrabuf里面，然后程序再把extrabuf里面的数据append到buffer中，如果读取数据比较少，那么数据直接存在buffer里面，不需要额外的操作。

buffer数据结构

prependable = readIndex
readable = writeIndex-readIndex
writable = size()-writeIndex

数据结构的初始化如下：

在数据的读取过程中，readIndex和writeIndex随着读取操作的进行而改变。

自动增长和内部腾挪

当当前的可写空间不够的时候，那么需要分配一个更大的内存，在分配好内存进行数据的复制时，顺便进行碎片回收，将所有数据负载到新内存的kCheapPrependable个字节后面。由于vector重新分配了内存，原有的指向其元素的指针会失效，所以buffer中采用的是readIndex和writeIndex来表示可读和可写的其实下标。

具体的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

void makeSpace(size_t len)
  {
    if (writableBytes() + prependableBytes() < len + kCheapPrepend)
    {
      // FIXME: move readable data
      buffer_.resize(writerIndex_+len);
    }
    else
    {
      // move readable data to the front, make space inside buffer
      assert(kCheapPrepend < readerIndex_);
      size_t readable = readableBytes();
      std::copy(begin()+readerIndex_,
                begin()+writerIndex_,
                begin()+kCheapPrepend);
      readerIndex_ = kCheapPrepend;
      writerIndex_ = readerIndex_ + readable;
      assert(readable == readableBytes());
    }
  }

前方添加

prependable的作用就是实现高效率的前方添加。预留一定的小空间，但需要在数据的前面添加数据时，就可以直接的添加了，否则的话，需要对整个buffer进行操作。
例如：程序以固定的4字节表示数据的长度，但需要序列化一个消息的时候，不知道该消息的长度，那么可以一直append直到序列化完成为止，然后就可以在前面添加最终的长度。

fork一般不能在多线程程序中调用，因为Linux的fork()只克隆当前线程的thread of control，不克隆其他的线程。fork()之后，除了当前的线程之外，其他的线程都消失。也就是说fork不能一下子得到一个与父进程一样的多线程子进程。

fork()之后，子进程中已有一个线程，其他的线程都消失了，这就造成一个危险的局面。在执行fork前，有一个mutex被线程锁住了，此时调用fork，子进程空间的mutex会处于锁住状态，由于父进程中的mutex与子进程中的muetx属于不同的空间，即使父进程中的mutex别解锁了，此时子进程中的mutex也是处于锁住状态，在子进程中根本就没有谁拥有该锁，这就导致子进程的mutex一直处于锁住状态，此时如果子进程对mutex加锁的话，那么就会进入死锁。

POSIX threads库提供了pthread_self接口，可以获得进程里面的线程id，其类型为pthread_t类型。pthread_t不一定是一个数值类型，也有可能是一个结构体，它是一个非可移植性的类型，也就是说，在这个系统中，可能是unsigned int类型，在别的系统可能是long，double，或者甚至就是个结构体，都不得而知，所以非可移植。因此Pthreads专门提供了pthread_equal函数用于比较两个线程标识符是否相等。这就带来了一系列问题：

1. 无法打印输出pthread_t，因为不知道其正真的类型。
2. 无法比较pthread_t的大小或是hash值，所以不能作为关联容器的key
3. pthread_t只是在进程内有意义，与操作系统的任务调度之间无法建立有效的关联。在/proc文件系统中找不到pthread_t对应的task。

因此，pthread_t并不适合作为程序中对线程的标识符。

在Linux上，可以使用gettid系统调用来获得线程的id，这么做的好处为：

1. 他的类型为pid_t, 其值通常是一个小的整数型，便于在日志中输出
2. 在现代的Linux操作系统中，她直接表示内核的任务调度id，因此在/proc文件系统中可以轻易找到其对应项
3. 在其他的系统工具中容易定位到具体的线程。例如在top命令中，可以按照线程列出任务，可以根据CPU或是内存的使用率定位到具体的线程
4. 任何时刻都是全局唯一的。
5. 0为非法值，因为操作系统第一个进程init的pid为1。

如何获取线程的TID（thread ID）?

通过查看man得到如下描述：

(1) The gettid() system call first appeared on Linux in kernel 2.4.11.
(2) gettid() returns the thread ID of the current process. This is equal to the process ID (as returned by getpid(2)),
unless the process is part of a thread group (created by specifying the CLONE_THREAD flag to the clone(2) system call). All processes in the same thread group have the same PID, but each one has a unique TID.
(3) gettid() is Linux specific and should not be used in programs that are intended to be portable.
（如果考虑移植性，不应该使用此接口）

1
2
3
4
5

#include <sys/syscall.h>  
#define gettidv1() syscall(__NR_gettid)  
#define gettidv2() syscall(SYS_gettid)  
printf("The ID of this thread is: %ld\n", (long int)gettidv1());// 最新的方式  
printf("The ID of this thread is: %ld\n", (long int)gettidv2());// 传统方式

单例模式的实现理论上采用double checked locking 不会出现线程安全问题，但是有可能编译器会对程序进行优化，使得程序乱序执行，导致出现错误。在Java中，可以借助类的转载阶段初始化静态区域来避免该问题，但是C++没有这个机制。在C++的单例实现中，可以使用pthread_once()函数来实现。

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

/*
 *在Linux thread中，一次性函数的执行有三种状态：NEVER(0), IN_PROGRESS(1), DONE(3),
 *处于IN_PROGRESS状态的话，所有所有调用 pthrad_once函数都会等待“已执行一次信号”才会退出阻塞，
 *对于处于DONE的话，所有pthread_once函数都会立刻返回0.所以_ponce的初始值
 *应该为PTHREAD_ONCE_INIT（值为0）
 */
template<typename T>
class Singleton: boost::nocopyable{
  public:
    static T& instance(){
      pthread_once(&ponce_, &Singleton::init);
      return *value_;
    }

  private:
    Singleton();
    ~Singleton();
    static void init(){
      value_ = new T(); //这里一定要使用new，使得对象存在对中，不会随着函数的退出而消亡
    }

  //静态成员变量只能在类外定义
  private:
    static pthread_once_t ponce_;
    static T* value_;
};

//必须在头文件里面定义static变量
template<typename T>
pthread_once_t Singleton<T>::ponce_ = PTHREAD_ONCE_INIT;

template<typename T>
T* Singleton<T>::value_ = nullptr;

条件变量的正确使用方式一般为：

1. 必须与mutex一起使用，该布尔表达式的读写需要受到mutex的保护
2. 在mutex已上锁的时候才能用 wait()
3. 吧判断布尔条件和wait()放到while循环中

写成代码为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

mudo::MutexLock mutex;
mudo::Condition cond(mutex);
std::dequeue<int> queue;

int dequeue(){
  MutexLockGuard lock(mutex);
  while(queue.empty()){
    cond.wait();  //会原子的unlock mutex进入等待，不会与enqueue形成死锁，当wait出来后，会自动加锁
  }
  assert(!queue.empty());
  int top = queue.front();
  queue.pop_front();
  return top;
}

使用while循环来等待条件变量，而不能使用if的原因就是在linux下有时会出现虚假唤醒，在多核的情况下，一个noify有可能会唤醒多个wait函数，造成虚假唤醒。还有一个原因就是，如果使用if语句的话，那么有可能在enqueue的时候不小心使用了broadcast，造成所有等待wait的线程被唤醒，每一个线程都加锁后进行pop操作，就有可能导致队列为空的情况下还在pop，从而出现错误。

对于signal和broadcast端：

1. 在signal之前一定要修改布尔表达式
2. 修改布尔表达式要在mutex的保护下
3. 注意区分signal和broadcast：broadcast通常用于表明状态变化，signal通常用于表示资源可用

代码为：

1
2
3
4
5

void enqueue(int x){
  MutexLockGuard lock(mutex);
  queue.push_back(x);
  cond.notify() //notify可以移出临界区之外
}

在上面的代码中，enqueue每一次push都会调用notify，能不能在队列从0变为1的时候才调用notify？

对于只有一个dequeue的线程的话可以这么做，但对于有多个dequeue的话，就不行。因为只有在0变1
的情况下调用notify，此时只有一个线程被唤醒，即使队列有很多资源可以使用，其他线程也会一直
阻塞在wait里面。造成线程浪费，程序并发性得不到应用。

DFS+ST算法可以在线的查询两个节点的最近公共祖先。由于任意两点的最近公共祖先必定在这两点的最短路径上，所以也可以根据这个特性来查找任意两点的最近路径。
DFS：深度优先遍历树的每一个节点，用数组deep按照遍历顺序记录下每一个节点的深度，另外需要记录每一个节点在数组中所对应的第一个出现的下标。
ST：有了深度数组，利用ST算法计算每一个偶数段的区间的最小深度的下标。
对于每一个查询(a, b), 先获得a和b在深度数组中第一次出现的下标firsta和firstb，接下来就是求出在深度数组中下标从firsta到firstb的最小值的下标，改下标对应的节点就是最近公共祖先。求最小值的下标ST算法可以在线的以常数的时间给出。如果需要求出两点之前的最短路径长度，那么可以直接的deep[a]+deep[b]-2*deep[最近公共祖先下标]。

例题：

采用DFS遍历整棵树，得到以下数据：

（1）遍历序列p：0　　1　　3　　1　　4　　7　　4　　8　　4　　1　　5　　1　　0　　2　　6　　2　　0

（2）各节点的深度序列　　　　　　　 depth: 0　　1　　 1　　 2　　 2　　2　　 2　　 3　　3

（3）各节点在序列p中首次出现的位置序列pos: 0　　1　　13　　2　　4　　10　　14　　5　　7

使用ST算法，假设现在我们要求节点7和5的最短路径，我们可以这样做：

（1）首先，从pos序列中获得节点7和节点5在p序列中第一次出现的位置分别为：pos[7] = 5, pos[5] = 10;

（2）得到p序列中[5, 10]这一段子序列s：7　　4　　8　　4　　1　　5

（3）s序列中深度最小的点即节点1就是我们要找的节点7和节点5的LCA。

例子：最近公共祖先·三

描述

上上回说到，小Hi和小Ho使用了Tarjan算法来优化了他们的“最近公共祖先”网站，但是很快这样一个离线算法就出现了问题：如果只有一个人提出了询问，那么小Hi和小Ho很难决定到底是针对这个询问就直接进行计算还是等待一定数量的询问一起计算。毕竟无论是一个询问还是很多个询问，使用离线算法都是只需要做一次深度优先搜索就可以了的。

那么问题就来了，如果每次计算都只针对一个询问进行的话，那么这样的算法事实上还不如使用最开始的朴素算法呢！但是如果每次要等上很多人一起的话，因为说不准什么时候才能够凑够人——所以事实上有可能要等上很久很久才能够进行一次计算，实际上也是很慢的！

“那到底要怎么办呢？在等到10分钟，或者凑够一定数量的人两个条件满足一个时就进行运算？”小Ho想出了一个折衷的办法。

“哪有这么麻烦！别忘了和离线算法相对应的可是有一个叫做在线算法的东西呢！”小Hi笑道。

小Ho面临的问题还是和之前一样：假设现在小Ho现在知道了N对父子关系——父亲和儿子的名字，并且这N对父子关系中涉及的所有人都拥有一个共同的祖先（这个祖先出现在这N对父子关系中），他需要对于小Hi的若干次提问——每次提问为两个人的名字（这两个人的名字在之前的父子关系中出现过），告诉小Hi这两个人的所有共同祖先中辈分最低的一个是谁？

提示：最近公共祖先无非就是两点连通路径上高度最小的点嘛！

输入

每个测试点（输入文件）有且仅有一组测试数据。

每组测试数据的第1行为一个整数N，意义如前文所述。

每组测试数据的第2~N+1行，每行分别描述一对父子关系，其中第i+1行为两个由大小写字母组成的字符串Father_i, Son_i，分别表示父亲的名字和儿子的名字。

每组测试数据的第N+2行为一个整数M，表示小Hi总共询问的次数。

每组测试数据的第N+3~N+M+2行，每行分别描述一个询问，其中第N+i+2行为两个由大小写字母组成的字符串Name1_i, Name2_i，分别表示小Hi询问中的两个名字。

对于100%的数据，满足N<=10^5，M<=10^5, 且数据中所有涉及的人物中不存在两个名字相同的人（即姓名唯一的确定了一个人），所有询问中出现过的名字均在之前所描述的N对父子关系中出现过，且每个输入文件中第一个出现的名字所确定的人是其他所有人的公共祖先。

输出

对于每组测试数据，对于每个小Hi的询问，按照在输入中出现的顺序，各输出一行，表示查询的结果：他们的所有共同祖先中辈分最低的一个人的名字。

样例输入

4
Adam Sam
Sam Joey
Sam Micheal
Adam Kevin
3
Sam Sam
Adam Sam
Micheal Kevin

样例输出

Sam
Adam
Adam

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

#include <iostream>
#include <string>
#include <vector>
#include <queue>
#include <map>
#include <algorithm>

using namespace std;

const int maxn = 200009;
const int maxl = 21;

map <string, int> rn;
vector <int> e[maxn];
string name[maxn];
int n, m, tn, ath[maxn][maxl], d[maxn];

int getNum(string n) {
	map <string, int> :: iterator i = rn. find(n);
	if (i == rn. end()) {
		name[++ tn] = n;
		rn. insert(pair<string, int> (n, tn));
		return tn;
	}
	else
		return i-> second;
}

void buildTree() {
	queue <int> q;
	int p0 = 1;
	while (ath[p0][0])
		p0 = ath[p0][0];
	d[p0] = 1;
	q. push(p0);
	while (!q. empty()) {
		int p = q. front();
		q. pop();
		for (int i = 1; i < maxl; ++ i)
			ath[p][i] = ath[ath[p][i - 1]][i - 1];
		for (vector <int> :: iterator i = e[p]. begin(); i != e[p]. end(); ++ i) {
			q. push(*i);
			d[*i] = d[p] + 1;
		}
	}
}

int LCA(int p, int q) {
	if (d[p] < d[q])
		swap(p, q);
	for (int i = maxl - 1; i >= 0; -- i)
		if (d[ath[p][i]] >= d[q])
			p = ath[p][i];
	if (p == q)
		return p;
	for (int i = maxl - 1; i >= 0; -- i)
		if (ath[p][i] ^ ath[q][i])
			p = ath[p][i], q = ath[q][i];
	return ath[p][0];
}

int main() {
	cin. sync_with_stdio(0);
	cin >> n;
	tn = 0;
	while (n --) {
		string a, b;
		cin >> a >> b;
		int na = getNum(a), nb = getNum(b);
		ath[nb][0] = na;
		e[na]. push_back(nb);
	}
	buildTree();
	cin >> m;
	while (m --) {
		string a, b;
		cin >> a >> b;
		int na = getNum(a), nb = getNum(b);
		cout << name[LCA(na, nb)] << endl;
	}
}

代码来自hiho一下 第十七周的排名第一参赛者laekov

寻找最近公共祖先有两种效率高的算法，一种是离线算法Tarjan算法，用户输入所有查询，算法给出所有查询结果，算法在运行的时候就已经知道用户的所有查询了，时间复杂度为O(nlogn)。另外一种是在线算法DFS+ST算法，算法运行不需要知道用户的输入，每一次查询可以在常数时间内给出结果。时间复杂度为DFS->O(n), ST->O(nlogn)。这里主要讲一下Tarjan算法。

Tarjan算法是DFS搜索和并查集的思想的结合，深度搜索每一个节点，对于当前的节点x，将其看做一个新的集合，并将其作为该集合的代表，查询处理与该节点有关的查询，接着递归遍历该节点的所有子节点，在遍历完所有子节点后，将该集合与root集合合并，也就是并查集的Union操作。在查询的过程中还可以使用路径压缩的方法来提高查询效率。例如：如果有查询（a, b）,那么当遍历到a节点的时候，如果b节点已经遍历过了，那么分两种情况，如果b在另外一个子树上，那么查找b的时候，b所在的结合的代表就是a和b的最近公共祖先。如果b不在另一颗子树上，而是与a同在一颗子树，那么此时的b就是最近公共祖先。这也就是为什么在需要在遍历完节点的所有子节点后才可以将当前节点与root合并的原因。

例子：Nearest Common Ancestors

Description

A rooted tree is a well-known data structure in computer science and engineering. An example is shown below:

In the figure, each node is labeled with an integer from {1, 2,…,16}. Node 8 is the root of the tree. Node x is an ancestor of node y if node x is in the path between the root and node y. For example, node 4 is an ancestor of node 16. Node 10 is also an ancestor of node 16. As a matter of fact, nodes 8, 4, 10, and 16 are the ancestors of node 16. Remember that a node is an ancestor of itself. Nodes 8, 4, 6, and 7 are the ancestors of node 7. A node x is called a common ancestor of two different nodes y and z if node x is an ancestor of node y and an ancestor of node z. Thus, nodes 8 and 4 are the common ancestors of nodes 16 and 7. A node x is called the nearest common ancestor of nodes y and z if x is a common ancestor of y and z and nearest to y and z among their common ancestors. Hence, the nearest common ancestor of nodes 16 and 7 is node 4. Node 4 is nearer to nodes 16 and 7 than node 8 is.

For other examples, the nearest common ancestor of nodes 2 and 3 is node 10, the nearest common ancestor of nodes 6 and 13 is node 8, and the nearest common ancestor of nodes 4 and 12 is node 4. In the last example, if y is an ancestor of z, then the nearest common ancestor of y and z is y.

Write a program that finds the nearest common ancestor of two distinct nodes in a tree.

Input

The input consists of T test cases. The number of test cases (T) is given in the first line of the input file. Each test case starts with a line containing an integer N , the number of nodes in a tree, 2<=N<=10,000. The nodes are labeled with integers 1, 2,…, N. Each of the next N -1 lines contains a pair of integers that represent an edge –the first integer is the parent node of the second integer. Note that a tree with N nodes has exactly N - 1 edges. The last line of each test case contains two distinct integers whose nearest common ancestor is to be computed.
Output

Print exactly one line for each test case. The line should contain the integer that is the nearest common ancestor.

Sample Input

2
16
1 14
8 5
10 16
5 9
4 6
8 4
4 10
1 13
6 15
10 11
6 7
10 2
16 3
8 1
16 12
16 7
5
2 3
3 4
3 1
1 5
3 5

Sample Output

4
3

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

#include<stdio.h>
#include<vector>
#include<string.h>
using namespace std;
#define Size 11111  //节点个数

vector<int> node[Size],que[Size];
int n,pare[Size],anse[Size],in[Size],rank[Size];

int vis[Size];
void init()
{
	int i;
	for(i=1;i<=n;i++)
	{
		node[i].clear();
		que[i].clear();
		rank[i]=1;
		pare[i]=i;///
	}
	memset(vis,0,sizeof(vis));
	memset(in,0,sizeof(in));
	memset(anse,0,sizeof(anse));

}

int find(int nd)
{
	return pare[nd]==nd?nd:pare[nd]=find(pare[nd]);
}
int Union(int nd1,int nd2)
{
	int a=find(nd1);
	int b=find(nd2);
	if(a==b) return 0;
	else if(rank[a]<=rank[b])
	{
		pare[a]=b;
		rank[b]+=rank[a];
	}
	else
	{
		pare[b]=a;
		rank[a]+=rank[b];
	}
	return 1;

}

void LCA(int root)
{
	int i,sz;
	anse[root]=root;//首先自成一个集合
	sz=node[root].size();
	for(i=0;i<sz;i++)
	{
		LCA(node[root][i]);//递归子树
		Union(root,node[root][i]);//将子树和root并到一块
		anse[find(node[root][i])]=root;//修改子树的祖先也指向root
	}
	vis[root]=1;
	sz=que[root].size();
	for(i=0;i<sz;i++){
      if(vis[que[root][i]]){

        //root和que[root][i]所表示的值的最近公共祖先
				printf("%d\n",anse[find(que[root][i])]);
				return ;
			}
	}
     return ;
}

int main()
{
	int cas,i;
	scanf("%d",&cas);
	while(cas--)
	{
		int s,e;
		scanf("%d",&n);
		init();
		for(i=0;i<n-1;i++)
		{
			scanf("%d %d",&s,&e);
			if(s!=e)
			{
			    node[s].push_back(e);
			   // node[e].push_back(s);
				in[e]++;
			}
		}
		scanf("%d %d",&s,&e);
		que[s].push_back(e);
		que[e].push_back(s);
		for(i=1;i<=n;i++)  if(in[i]==0) break;//寻找根节点
	  //	printf("root=%d\n",i);
		LCA(i);
	}
	return 0;
}

代码转载自这里

例子：CD操作

Problem Description

在Windows下我们可以通过cmd运行DOS的部分功能，其中CD是一条很有意思的命令，通过CD操作，我们可以改变当前目录。
这里我们简化一下问题，假设只有一个根目录，CD操作也只有两种方式： 　　

1. CD 当前目录名...\目标目录名 (中间可以包含若干目录，保证目标目录通过绝对路径可达)
2. CD .. (返回当前目录的上级目录)
   　　
   现在给出当前目录和一个目标目录，请问最少需要几次CD操作才能将当前目录变成目标目录？

Input

输入数据第一行包含一个整数T(T<=20)，表示样例个数；
每个样例首先一行是两个整数N和M(1<=N,M<=100000)，表示有N个目录和M个询问；
接下来N-1行每行两个目录名A B(目录名是只含有数字或字母，长度小于40的字符串)，表示A的父目录是B。
最后M行每行两个目录名A B，表示询问将当前目录从A变成B最少要多少次CD操作。
数据保证合法，一定存在一个根目录，每个目录都能从根目录访问到。

Output

请输出每次询问的结果，每个查询的输出占一行。

Sample Input

2
3 1
B A
C A
B C

3 2
B A
C B
A C
C A

Sample Output

2
1
2

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

//思路：

//　　求出a和b的最近公共祖先,然后分4种情况讨论

//　　①. a和b有一个公共祖先c，则用 c时间戳-a的时间戳+1(1步可以直接从c到b)

//　　②. a是b的祖先，则只用1步就可以到达b点

//　　③. b是a的祖先，则用a的时间戳-b的时间戳

//　　④. a和b是同一个点，则答案是0

#include<stdio.h>
#include<vector>
#include<string.h>
#include<map>
#include<math.h>
#include<string>
using namespace std;
#define Size 111111  //节点个数
struct Query
{
    int nd,id;
}temp;
struct out
{
    int s,e;
}out[Size];
vector<int> node[Size];
vector<struct Query>que[Size];
int n,m,pare[Size],ance[Size],in[Size],rank[Size],dis[Size],ans[Size],vis[Size];
map<string,int>mp;
void init()
{
	int i;
	for(i=1;i<=n;i++)
	{
		node[i].clear();
		que[i].clear();
		rank[i]=1;
		pare[i]=i;///
	}
	memset(vis,0,sizeof(vis));
	memset(in,0,sizeof(in));
	memset(ance,0,sizeof(ance));
	memset(dis,0,sizeof(dis));
	mp.clear();
}
int aabs(int aa)
{
     if(aa>0) return aa;
     else return -aa;
}
int find(int nd)//并查集操作  不解释
{
	return pare[nd]==nd?nd:pare[nd]=find(pare[nd]);
}
int Union(int nd1,int nd2)//并查集操作  不解释
{
	int a=find(nd1);
	int b=find(nd2);
	if(a==b) return 0;
	else if(rank[a]<=rank[b])
	{
		pare[a]=b;
		rank[b]+=rank[a];
	}
	else
	{
		pare[b]=a;
		rank[a]+=rank[b];
	}
	return 1;
}
void LCA(int root,int num)
{
	int i,sz;
	ance[root]=root;//首先自成一个集合
	dis[root]=num;
	sz=node[root].size();
	for(i=0;i<sz;i++)
	{
		   LCA(node[root][i],num+1);//递归子树
		   Union(root,node[root][i]);//将子树和root并到一块
		 ance[find(node[root][i])]=root;//修改子树的祖先也指向root
	}
	vis[root]=1;
	sz=que[root].size();
	for(i=0;i<sz;i++)
	{
	    int nd1,nd2,idx,ancestor;
	    nd1=root;nd2=que[root][i].nd;idx=que[root][i].id;
            if(vis[nd2])
			{
                  ans[idx]=ance[find(nd2)];
			}
	}
     return ;
}

int main()
{
	int cas,i;
	scanf("%d",&cas);
	while(cas--)
	{
		char  ss[100],ee[100];
		int s,e,cnt=1;
		scanf("%d %d",&n,&m);
		init();
		for(i=0;i<n-1;i++)
		{
			scanf("%s %s",ee,ss);
			if(mp.find(ss)==mp.end())
            {
                 s=cnt;mp[ss]=cnt++;
            }
            else s=mp[ss];
			if(mp.find(ee)==mp.end())
            {
                e=cnt;mp[ee]=cnt++;
            }
            else  e=mp[ee];
			if(s!=e)
			{
			    node[s].push_back(e);
				in[e]++;
			}
		}
		for(i=0;i<m;i++)
        {
		   scanf("%s %s",ss,ee);
		   s=mp[ss];e=mp[ee];
		   out[i].s=s;out[i].e=e;
		   temp.nd=e;temp.id=i;
		   que[s].push_back(temp);
		   temp.nd=s;temp.id=i;
		   que[e].push_back(temp);
        }
		for(i=1;i<=n;i++)  if(in[i]==0) break;//寻找根节点
		LCA(i,0);
		for(i=0;i<m;i++)
        {
            if(out[i].s==out[i].e)
                printf("0\n");
            else
                if(out[i].s==ans[i])
                      printf("1\n");
            else if(out[i].e==ans[i])
                printf("%d\n",dis[out[i].s]-dis[ans[i]]);
            else
            printf("%d\n",dis[out[i].s]-dis[ans[i]]+1);
        }
	}
	return 0;
}

代码转载自这里

例子：Connections between cities

Problem Description

After World War X, a lot of cities have been seriously damaged, and we need to rebuild those cities. However, some materials needed can only be produced in certain places. So we need to transport these materials from city to city. For most of roads had been totally destroyed during the war, there might be no path between two cities, no circle exists as well.
Now, your task comes. After giving you the condition of the roads, we want to know if there exists a path between any two cities. If the answer is yes, output the shortest path between them.

Input

Input consists of multiple problem instances.For each instance, first line contains three integers n, m and c, 2<=n<=10000, 0<=m<10000, 1<=c<=1000000. n represents the number of cities numbered from 1 to n. Following m lines, each line has three integers i, j and k, represent a road between city i and city j, with length k. Last c lines, two integers i, j each line, indicates a query of city i and city j.

Output

For each problem instance, one line for each query. If no path between two cities, output “Not connected”, otherwise output the length of the shortest path between them.

Sample Input

5 3 2
1 3 2
2 4 3
5 2 3
1 4
4 5

Sample Output

Not connected
6

Hint

Huge input, scanf recommended.

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

#include<stdio.h>
#include<string.h>
#include<vector>
using namespace std;
#define Size  11111
struct Edge
{
    int y,val;
}temp;
struct Query
{
    int y,id;
}mid;
int pare[Size],ance[Size],vis[Size],dis[Size],rank[Size],ans[1000000+100],n,m,c,tree[Size];
vector<struct Query>que[Size];
vector<struct Edge>node[Size];
void init()
{
    int i;
    for(i=0;i<=n;i++)
    {
        vis[i]=0;
        pare[i]=i;
        dis[i]=0;
        rank[i]=1;
        que[i].clear();
        node[i].clear();
    }
    memset(ans,-1,sizeof(ans));
}
int find(int x)
{
    return pare[x]==x?x:pare[x]=find(pare[x]);
}
/*
void Union(int x,int y)
{
x=find(x);
y=find(y);
if(x!=y)
{
if(rank[x]>rank[y])
{
rank[x]+=rank[y];
pare[y]=x;
}
else
{
rank[y]+=rank[x];
pare[x]=y;
}
}
}
*/
void LCA(int root,int d,int k)//k表示是以第k个点作为根的树
{
    int i,sz,nd1,nd2;
    vis[root]=1; //已经遍历过的点 要标记一下 不要
    tree[root]=k;dis[root]=d;
    // ance[root]=root;
    sz=node[root].size();
    for(i=0;i<sz;i++)
    {
        nd2=node[root][i].y;
        if(!vis[nd2])
        {
            LCA(nd2,d+node[root][i].val,k);
            // Union(node[root][i].y,root);//用带rank的幷查集操作答案不对 不知道why
            int w=find(nd2),m=find(root);
            if(w!=m)
            {
               pare[w]=m;//这样才对
            }
            //ance[find(node[root][i].y)]=root;
        }
    }
    sz=que[root].size();
    for(i=0;i<sz;i++)
    {
        nd1=root;
        nd2=que[root][i].y;
        if(vis[nd2]&&tree[nd1]==tree[nd2])//如果 nd1 nd2 的跟是同一个点 则是同一棵树上的
        {
            ans[que[root][i].id]=dis[nd1]+dis[nd2]-2*dis[find(nd2)];
        }
    }
}
int main()
{
    int i,j,x,y,val;
    while(scanf("%d %d %d",&n,&m,&c)!=EOF)
    {
        init();
        for(i=0;i<m;i++)
        {
            scanf("%d %d %d",&x,&y,&val);
            if(x!=y)
            {
                temp.y=y;temp.val=val;
                node[x].push_back(temp);
                temp.y=x;
                node[y].push_back(temp);//路是2个方向都可以通行的
            }
        }
        for(i=0;i<c;i++)
        {
            scanf("%d %d",&x,&y);
            mid.id=i;
            mid.y=y;
            que[x].push_back(mid);
            mid.y=x;
            que[y].push_back(mid);
        }
        for(i=1;i<=n;i++)
        {
            LCA(i,0,i);//以每一个节点作为根节点去深度搜索  找出每个点作为根的所有最近公共祖先
        }
        for(i=0;i<c;i++)
        {
            if(ans[i]==-1)
                printf("Not connected\n");
            else
                printf("%d\n",ans[i]);
        }
    }
    return 0;
}
//本题给的是一个森林 而不是一颗树，由于在加入边的时候，我们让2个方向都能走这样就形成了一个强连通的快，
//对于这个快来说，不管从快上那点出发 都可以遍历这个快上的所有的点，且相对距离是一样的

代码转载自这里

ST（Sparse Table）算法是一个非常有名的在线处理RMQ问题的算法，它可以在O(nlogn)时间内进行预处理，然后在O(1)时间内回答每个查询。

ST（Sparse Table）算法是基于动态规划的，用f[i][j]表示区间起点为j长度为2^i的区间内的最小值所在下标，通俗的说，就是区间[j, j + 2^i)的区间内的最小值的下标。

从定义可知，这种表示法的区间长度一定是2的幂，所以除了单位区间(长度为1的区间)以外，任意一个区间都能够分成两份，并且同样可以用这种表示法进行表示，[j, j + 2^i)的区间可以分成[j, j+2^(i-1))和[j + 2^i)，于是可以列出状态转移方程为： f[i][j] = RMQ( f[i-1][j], f[i-1][j+2^(i-1)] )。

f数组记录的是长度为偶数的子串的最值，对于正常的查询的话，并不可能每一次都是查询偶数长的区间，所以还需要在进一步的处理一下。对于查询区间[a, b], 可以将其划分为两个长度相等的偶数的子区间[a, a+2^k],[b-2^k, b],这两个区间有可能是相交的，但是不影响对区间最值的求解。现在只要根据a、b求出k，就可以知道[a, b]区间的最值了，为：min/max{f[k][a], f[k][b-(1<<k)+1]}.

对于K，需要满足a+2^k-1 >= b-2^k，则2^(k+1) >= (b-a+1), 两边取对数（以2为底），得 k+1 >= lg(b-a+1)，则k >= lg(b-a+1) - 1，k只要需要取最小的满足条件的整数即可。

初始化：

1
2
3
4
5
6
7
8
9
10

//n为元数的个数
//bitn为n的二进制位数，取下整(int)(log(n)/log(2))
for (int i=0; i<n; ++i)
   f[i][0]=input[i];
for (int j=1; j<bitn; ++j)
   for (int i=0; i<n; ++i)
   {
       if (i+(1<<(j-1))>=n) break;
       f[i][j]=max(f[i][j-1],f[i+(1<<(j-1))][j-1]);
    }

查询：

1
2
3
4
5

int query(int s,int e)  //查询区间[s,e]的最值
{
  int k=(int)((log(e-s+1.0)/log(2.0)));
  return max(f[s][k],f[e-(1<<k)+1][k]);
}

是一种二叉搜索树，它将一个区间划分成一些单元区间，每个单元区间对应线段树中的一个叶结点。
对于线段树中的每一个非叶子节点[a,b]，它的左子树表示的区间为[a,(a+b)/2]，右子树表示的区间为[(a+b)/2+1,b]。因此线段树是平衡二叉树。叶节点数目为N，即整个线段区间的长度。
线段树主要的操作有建树、更新节点、查找三种操作

build

线段树使用数组来表示二叉树，其建树代码如下：

1
2
3
4
5
6
7
8

#define lchild rt << 1, l, m
#define rchild rt << 1 | 1, m + 1, r
void build(int rt = 1, int l = 1, int r = N) {
    if (l == r) { std::cin >> tree[rt]; return; }
    int m = (l + r) >> 1;
    build(lchild); build(rchild);
    push_up(rt);
}

节点数据向上更新

线段树的作用有很多种，可以表示区间的最大值或是最小值，也可以表示区间的和等，所以具体的更新操作update或是push_up操作需要根据功能来决定。下面是两种不用功能时的push_up操作：

1
2
3
4
5
6
7
8
9

/* 对于区间求和 */
void push_up(int rt) {
    tree[rt] = tree[rt << 1] + tree[rt << 1 | 1];
}

/* 对于区间求最大值 */
void push_up(int rt) {
    tree[rt] = max(tree[rt << 1], tree[rt << 1 | 1]);
}

节点懒惰标记下推

对于区间求和, 原子数组值需要加上lazy标记乘以子树所统计的区间长度。 len为父节点统计的区间长度, 则len - (len >> 1)为左子树区间长度, len >> 1为右子树区间长度。

1
2
3
4
5
6
7

void push_down(int rt, int len) {
    tree[rt << 1] += lazy[rt] * (len - (len >> 1));
    lazy[rt << 1] += lazy[rt];
    tree[rt << 1 | 1] += lazy[rt] * (len >> 1);
    lazy[rt << 1 | 1] += lazy[rt];
    lazy[rt] = 0;
}

对于区间求最大值, 子树的值不需要乘以长度, 所以不需要传递参数len。

1
2
3
4
5
6
7

void push_down(int rt) {
    tree[rt << 1] += lazy[rt];
    lazy[rt << 1] += lazy[rt];
    tree[rt << 1 | 1] += lazy[rt];
    lazy[rt << 1 | 1] += lazy[rt];
    lazy[rt] = 0;
}

update

update操作也是根据具体功能来决定，下面的代码为单点更新：

1
2
3
4
5
6
7
8
9
10
11
12

#define lchild rt << 1, l, m
#define rchild rt << 1 | 1, m + 1, r
void update(int p, int delta, int rt = 1, int l = 1, int r = N) {
    if (l == r) {
        tree[rt] += delta;
        return;
    }
    int m = (l + r) >> 1;
    if (p <= m) update(p, delta, lchild);
    else update(p, delta, rchild);
    push_up(rt);
}

成段更新，需要用到lazy来提高效率,下面代码是区间和的情况：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

#define lchild rt << 1, l, m
#define rchild rt << 1 | 1, m + 1, r
void update(int L, int R, int delta, int rt = 1, int l = 1, int r = N) {
    if (L <= l && r <= R) {
        tree[rt] += delta * (r - l + 1);
        lazy[rt] += delta;
        return;
    }
    if (lazy[rt]) push_down(rt, r - l + 1);
    int m = (l + r) >> 1;
    if (L <= m) update(L, R, delta, lchild);
    if (R > m)  update(L, R, delta, rchild);
    push_up(rt);
}

query

查询也是要根据具体功能来实现，下面是查询区间和的代码

1
2
3
4
5
6
7
8
9
10

#define lchild rt << 1, l, m
#define rchild rt << 1 | 1, m + 1, r
int query(int L, int R, int rt = 1, int l = 1, int r = N) {
    if (L <= l && r <= R) return tree[rt];
    if (lazy[rt]) push_down(rt, r - l + 1);
    int m = (l + r) >> 1, ret = 0;
    if (L <= m) ret += query(L, R, lchild);
    if (R > m)  ret += query(L, R, rchild);
    return ret;
}

例子：hdu 1394 Minimum Inversion Number（线段树）

Problem Description

The inversion number of a given number sequence a1, a2, …, an is the number of pairs (ai, aj) that satisfy i < j and ai > aj.

For a given sequence of numbers a1, a2, …, an, if we move the first m >= 0 numbers to the end of the seqence, we will obtain another sequence. There are totally n such sequences as the following:

a1, a2, …, an-1, an (where m = 0 - the initial seqence)
a2, a3, …, an, a1 (where m = 1)
a3, a4, …, an, a1, a2 (where m = 2)
…
an, a1, a2, …, an-1 (where m = n-1)

You are asked to write a program to find the minimum inversion number out of the above sequences.

Input

The input consists of a number of test cases. Each case consists of two lines: the first line contains a positive integer n (n <= 5000); the next line contains a permutation of the n integers from 0 to n-1.

Output

For each case, output the minimum inversion number on a single line.

Sample Input

10
1 3 6 9 0 8 5 7 4 2

Sample Output

16

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

#include <cstdio>
#include <algorithm>
using namespace std;

#define lson l , m , rt << 1
#define rson m + 1 , r , rt << 1 | 1
const int maxn = 5555;
int sum[maxn<<2];
void PushUP(int rt) {
	sum[rt] = sum[rt<<1] + sum[rt<<1|1];
}
void build(int l,int r,int rt) {
	sum[rt] = 0;
	if (l == r) return ;
	int m = (l + r) >> 1;
	build(lson);
	build(rson);
}
void update(int p,int l,int r,int rt) {
	if (l == r) {
		sum[rt] ++;
		return ;
	}
	int m = (l + r) >> 1;
	if (p <= m) update(p , lson);
	else update(p , rson);
	PushUP(rt);
}
int query(int L,int R,int l,int r,int rt) {
	if (L <= l && r <= R) {
		return sum[rt];
	}
	int m = (l + r) >> 1;
	int ret = 0;
	if (L <= m) ret += query(L , R , lson);
	if (R > m) ret += query(L , R , rson);
	return ret;
}
int x[maxn];
int main() {
	int n;
	while (~scanf("%d",&n)) {
		build(0 , n - 1 , 1);
		int sum = 0;
		for (int i = 0 ; i < n ; i ++) {
			scanf("%d",&x[i]);
			sum += query(x[i] , n - 1 , 0 , n - 1 , 1);
			update(x[i] , 0 , n - 1 , 1);
		}
		int ret = sum;
		for (int i = 0 ; i < n ; i ++) {
			sum += n - x[i] - x[i] - 1;
			ret = min(ret , sum);
		}
		printf("%d\n",ret);
	}
	return 0;
}

例子：Billboard

Problem Description

At the entrance to the university, there is a huge rectangular billboard of size h*w (h is its height and w is its width). The board is the place where all possible announcements are posted: nearest programming competitions, changes in the dining room menu, and other important information.

On September 1, the billboard was empty. One by one, the announcements started being put on the billboard.

Each announcement is a stripe of paper of unit height. More specifically, the i-th announcement is a rectangle of size 1 * wi.

When someone puts a new announcement on the billboard, she would always choose the topmost possible position for the announcement. Among all possible topmost positions she would always choose the leftmost one.

If there is no valid location for a new announcement, it is not put on the billboard (that’s why some programming contests have no participants from this university).

Given the sizes of the billboard and the announcements, your task is to find the numbers of rows in which the announcements are placed.

Input

There are multiple cases (no more than 40 cases).

The first line of the input file contains three integer numbers, h, w, and n (1 <= h,w <= 10^9; 1 <= n <= 200,000) - the dimensions of the billboard and the number of announcements.

Each of the next n lines contains an integer number wi (1 <= wi <= 10^9) - the width of i-th announcement.

Output

For each announcement (in the order they are given in the input file) output one number - the number of the row in which this announcement is placed. Rows are numbered from 1 to h, starting with the top row. If an announcement can’t be put on the billboard, output “-1” for this announcement.

Sample Input

3 5 5
2
4
3
3
3

Sample Output

1
2
1
3
-1

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

#include <cstdio>
#include <algorithm>
using namespace std;

#define lson l , m , rt << 1
#define rson m + 1 , r , rt << 1 | 1
const int maxn = 222222;
int h , w , n;
int MAX[maxn<<2];
void PushUP(int rt) {
	MAX[rt] = max(MAX[rt<<1] , MAX[rt<<1|1]);
}
void build(int l,int r,int rt) {
	MAX[rt] = w;
	if (l == r) return ;
	int m = (l + r) >> 1;
	build(lson);
	build(rson);
}
int query(int x,int l,int r,int rt) {
	if (l == r) {
		MAX[rt] -= x;
		return l;
	}
	int m = (l + r) >> 1;
	int ret = (MAX[rt<<1] >= x) ? query(x , lson) : query(x , rson);
	PushUP(rt);
	return ret;
}
int main() {
	while (~scanf("%d%d%d",&h,&w,&n)) {
		if (h > n) h = n;
		build(1 , h , 1);
		while (n --) {
			int x;
			scanf("%d",&x);
			if (MAX[1] < x) puts("-1");
			else printf("%d\n",query(x , 1 , h , 1));
		}
	}
	return 0;
}

例子：Just a Hook

Problem Description

In the game of DotA, Pudge’s meat hook is actually the most horrible thing for most of the heroes. The hook is made up of several consecutive metallic sticks which are of the same length.

Now Pudge wants to do some operations on the hook.

Let us number the consecutive metallic sticks of the hook from 1 to N. For each operation, Pudge can change the consecutive metallic sticks, numbered from X to Y, into cupreous sticks, silver sticks or golden sticks.
The total value of the hook is calculated as the sum of values of N metallic sticks. More precisely, the value for each kind of stick is calculated as follows:

For each cupreous stick, the value is 1.
For each silver stick, the value is 2.
For each golden stick, the value is 3.

Pudge wants to know the total value of the hook after performing the operations.
You may consider the original hook is made up of cupreous sticks.

Input

The input consists of several test cases. The first line of the input is the number of the cases. There are no more than 10 cases.
For each case, the first line contains an integer N, 1<=N<=100,000, which is the number of the sticks of Pudge’s meat hook and the second line contains an integer Q, 0<=Q<=100,000, which is the number of the operations.
Next Q lines, each line contains three integers X, Y, 1<=X<=Y<=N, Z, 1<=Z<=3, which defines an operation: change the sticks numbered from X to Y into the metal kind Z, where Z=1 represents the cupreous kind, Z=2 represents the silver kind and Z=3 represents the golden kind.

Output

For each case, print a number in a line representing the total value of the hook after the operations. Use the format in the example.

Sample Input

1
10
2
1 5 2
5 9 3

Sample Output

Case 1: The total value of the hook is 24.

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

#include <cstdio>
#include <algorithm>
using namespace std;

#define lson l , m , rt << 1
#define rson m + 1 , r , rt << 1 | 1
const int maxn = 111111;
int h , w , n;
int col[maxn<<2];
int sum[maxn<<2];
void PushUp(int rt) {
	sum[rt] = sum[rt<<1] + sum[rt<<1|1];
}
void PushDown(int rt,int m) {
	if (col[rt]) {
		col[rt<<1] = col[rt<<1|1] = col[rt];
		sum[rt<<1] = (m - (m >> 1)) * col[rt];
		sum[rt<<1|1] = (m >> 1) * col[rt];
		col[rt] = 0;
	}
}
void build(int l,int r,int rt) {
	col[rt] = 0;
	sum[rt] = 1;
	if (l == r) return ;
	int m = (l + r) >> 1;
	build(lson);
	build(rson);
	PushUp(rt);
}
void update(int L,int R,int c,int l,int r,int rt) {
	if (L <= l && r <= R) {
		col[rt] = c;
		sum[rt] = c * (r - l + 1);
		return ;
	}
	PushDown(rt , r - l + 1);
	int m = (l + r) >> 1;
	if (L <= m) update(L , R , c , lson);
	if (R > m) update(L , R , c , rson);
	PushUp(rt);
}
int main() {
	int T , n , m;
	scanf("%d",&T);
	for (int cas = 1 ; cas <= T ; cas ++) {
		scanf("%d%d",&n,&m);
		build(1 , n , 1);
		while (m --) {
			int a , b , c;
			scanf("%d%d%d",&a,&b,&c);
			update(a , b , c , 1 , n , 1);
		}
		printf("Case %d: The total value of the hook is %d.\n",cas , sum[1]);
	}
	return 0;
}