Keep learning, keep living...

0%

TokyoCabinet(TC)提供了6种不同结构的数据库,包括:

  • (MDB) on-memory hash database
  • (NDB) on-memory tree database
  • (HDB) hash database
  • (BDB) B+ tree database
  • (FDB) fixed-length database
  • (TDB) table database

每种数据库都有各自一套API来进行各种操作。

为了简化使用,TC还提供了一套通用的API来操作以上所有类型数据库,叫做Abstract Database API.

Abstract Database API通过数据库名称来区分各类型数据库:

  • “*” on-memory hash database
  • “+” on-memory tree database
  • “.tch” hash database
  • “.tcb” B+ tree database
  • “.tcf” fixed-length database
  • “.tct” table database

不仅如此,TC更进一步进行了抽象,在Abstract Database中还提供了一种Skeleton Database。
通过实现Skeleton Database指定的API,可以使用自定义的数据库类型。
Skeleton Database API结构体如下:

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
typedef struct {                        /* type of structure for a extra database skeleton */
void *opq; /* opaque pointer */
void (*del)(void *); /* destructor */
bool (*open)(void *, const char *);
bool (*close)(void *);
bool (*put)(void *, const void *, int, const void *, int);
bool (*putkeep)(void *, const void *, int, const void *, int);
bool (*putcat)(void *, const void *, int, const void *, int);
bool (*out)(void *, const void *, int);
void *(*get)(void *, const void *, int, int *);
int (*vsiz)(void *, const void *, int);
bool (*iterinit)(void *);
void *(*iternext)(void *, int *);
TCLIST *(*fwmkeys)(void *, const void *, int, int);
int (*addint)(void *, const void *, int, int);
double (*adddouble)(void *, const void *, int, double);
bool (*sync)(void *);
bool (*optimize)(void *, const char *);
bool (*vanish)(void *);
bool (*copy)(void *, const char *);
bool (*tranbegin)(void *);
bool (*trancommit)(void *);
bool (*tranabort)(void *);
const char *(*path)(void *);
uint64_t (*rnum)(void *);
uint64_t (*size)(void *);
TCLIST *(*misc)(void *, const char *, const TCLIST *);
bool (*putproc)(void *, const void *, int, const void *, int, TCPDPROC, void *);
bool (*foreach)(void *, TCITER, void *);
} ADBSKEL;

各成员与其它类型API相应成员意义一致。在开发时,只需实现功能必需的相应函数,忽略其他成员。

使用示例:

1
2
3
4
5
6
7
8
9
10
ADBSKEL skel;
memset(0, &skel, sizeof(skel));
skel.opq = mydbnew();
skel.del = mydbdel;
skel.open = mydbopen;
skel.close = mydbclose;
...
TCADB *adb = tcadbnew();
tcadbsetskel(adb, &skel);
tcadbopen(adb, "foobarbaz");

为了解决多进程共享访问和远程访问TC数据库的不便与繁琐,TC作者开发了一个网络访问层,叫做TokyoTyrant(TT)。它使用TC的Abstract Database API来访问TC数据库。因而内置支持skeleton database扩展。
TT提供了-skel命令行选项来指定skeleton database,启动时它会加载传入的Shared Object(SO)文件,使用SO中定制的数据库实现。

1
ttserver -skel ttskelfoo.so

我们可以根据需求实现特定的SO文件,就可以完整利用TT本身已经实现的各种特性,如主备同步,memcache协议支持,HTTP协议支持等。在性能满足需求的情况,这将大大减少开发量。SO文件必须导出一个名字为initialize的函数,TT启动时会从SO文件中查找该函数来初始化skeleton database。
该函数原型为:

1
bool (*initfunc)(ADBSKEL *);

该函数传入一个指向skeleton database的指针。initialize函数中需要将skeleton database定制的数据库操作的API实现赋值到相应函数指针。
由于initialize函数没有参数传递TT本身相关信息,如命令行选项,配置结构等,而TT将一些信息存储在全局变量g_serv指向的TTSERV结构体中,因而SO中可以声明g_serv外部变量来引用。

1
extern TTSERV  *g_serv;

不过较为遗憾的是TTSERV中的信息较少,如有需要的话可以自行扩展。如果SO中逻辑需要依赖命令行选项,可以通过使用启动TT时传入的数据库名来做不同处理。skeleton database的open函数会传入该参数。

具体例子可以参考:
https://github.com/flygoast/ttskeliplist

opentracker是一个开源P2P tracker服务器.之前我们系统中主要使用的是PHP+MYSQL实现的peertracker。随着业务增长,peertracker的性能已经不能满足系统需要。因而我们决定引入性能更好的opentracker。不过, opentracker在功能上并不能完全满足我们的需求,因而我对它进行了一些扩展。

  • UDP单播同步数据

opentracker本身支持cluster模式。cluster内各节点之间会同步数据。这样可以通过添加节点提高整体的集群性能。然而,opentracker原生通过UDP多播进行数据同步。我们不具备多播IP,因而开发了单播模式进行同步。实现非常简单,就是依次向cluster内其他节点发送数据。缺点是当节点数较多时,会影响性能。

  • 持久化支持

opentracker将torrent和peer信息保存在内存中。当opentracker重启时,所有的torrent和peer信息就都丢失了。这会导致我们的系统一段时间内不能进行正常的P2P传输。因而我扩展了持久化功能。opentracker架构上通过多个不同的线程执行不同的任务。我添加了一个线程,周期性地将内存中的torrent和peer信息保存到磁盘文件中。这个线程很像Redis中进行数据dump的进程。磁盘文件格式定义为ODB(opentracker database),它主要借鉴自Redis的RDB格式。目前,没有处理IPV6格式,因而只支持IPV4。
我还提供了一个工具支持流式地对odb文件进行处理。

具体用法请参考: perldoc OdbParser

格式规范:

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
----------------------------------- # ODB is a binary format. There are no new lines or spaces in the file.
4f 50 45 4e 54 52 41 43 4b 45 52 # Magic String "OPENTRACKER"
30 30 30 33 # ODB Version Number in ASCII characters. In this case, version = "0001" = 1
-----------------------------------
FE # FE = Opcode that indicates following is a torrent information.
----------------------------------- # Torrent information starts from here.
00 02 2e 33 01 c4 a1 df bb 82
1f 51 0f b5 b6 02 6f 93 6e 9f # 20 byte info_hash of torrent
-----------------------------------
e2 37 59 01 00 00 00 00 # Last access time in minutes since UNIX epoch, 8 bytes.
# At present, when loading ODB file, just use current clock, ignore this.
-----------------------------------
00 00 00 00 00 00 00 00 # Seeding peer count for current torrent. 8 bytes long integer in little endian.
# NOT used at present.
-----------------------------------
00 00 00 00 00 00 00 00 # Total peer count for current torrent. 8 bytes long integer in little endian.
# NOT used at present.
-----------------------------------
00 00 00 00 00 00 00 00 # Download times of files in current torrent. 8 bytes long integer in little endian.
# NOT used at present.
-----------------------------------
01 00 00 00 # Peer count in current peer, 4 bytes integer in little endian.
----------------------------------- # Peers information starts from here.
7f 00 00 01 # 4 bytes ip address in network byte order. In this case, 0x7f000001 = "127.0.0.1"
1b 31 # 2 bytes port in network byte order. In this case, 0x1b31 = 6961.
80 # Flag of peers. SEEDING = 0x80, COMPLETED = 0x40, STOPPED = 0x20, LEECHING = 0x00
00 # Reserved. Just set zero.
-----------------------------------
... # Other peers information.
-----------------------------------
FE
-----------------------------------
... # Other torrent information.
-----------------------------------
FF # EOF opcode.
  • HTTP debug接口

由于opentracker响应内容是按BENCODE编码过的,调试时不太方便。因而扩展了一个返回human-readable内容的调试接口。

具体参看: https://github.com/flygoast/opentracker

我们线上服务使用nginx+passenger-3.0.11来运行Rails程序。我们发现当执行若干次nginx -s reload或者kill -HUP cat /usr/local/nginx/logs/nginx.pid之后,再执行以上命令不再生效。

用pstack观察master进程,master进程阻塞在read()操作上。

使用gcore生成core文件,然后用gdb查看,查看栈帧所在文件及行号。

查看passenger源码。readExact()逻辑为一直读到size大小返回。readArrayMessage()代码逻辑为先从传入fd中读取两个字节做为长度,再从fd中读取相应长度的内容。Passenger::AgentsStarter::start函数的过程为创建一个socketpair,然后fork()子进程,子进程执行PassengerWatchdog。父子进程通过该socketpair进行通信。master就是卡在读取socketpair上。

从gdb中看到readExact()传入的size为12848,即从socketpair中读到的长度为12848,而实际读取的内容长度为621,并且读取的内容也很诡异。应该是某个地方向socketpair中写入了错误内容。

继续查看passenger代码中子进程的逻辑。关闭socketpair一端。然后将socketpair另一端通过dup2()调用复制到3上。然后关闭除了0-3以外的所有文件描述符。接着解除信号阻塞,重置信号处理函数。

通过strace追踪master进程及其子进程行为,发现子进程确实发送了错误内容。”20”的十六进制表示为0x3230,按网络字节序读出正好是12848。从调用位置看是在SIGCHLD的信号处理函数中。而从写入内容看像是nginx在写日志。

继续追查为什么会有SIGCHLD产生。最后发现getHighestFileDescriptor()函数中子进程会fork()出孙进程,获取当前打开的最大fd,子进程等待孙进程退出后返回。由于孙进程退出,子进程收到了SIGCHLD信号。但是由于子进程是由master进程fork()出来的,SIGCHLD信号是被阻塞的。当执行sigprocmask()时,被阻塞的SIGCHLD被处理了,而这时的信号处理函数是ngx_signal_handler(), 该函数会调用ngx_log_error()向error log写入日志。若error log的fd为3,就发生我们上述的情况。

为了避免这种情况,应该先重置信号处理函数,再解除信号阻塞。

patch:

https://github.com/flygoast/passenger/commit/0b62808943aba65432f0b492f4ef941499fad02c

异常现象:

在/etc/rc.local中添加/usr/local/nginx/sbin/nginx来开机自动启动NGINX时,PassengerHelperAgent进程不停反复重启,而从shell上手动启动NGINX时一切正常。

追查过程:

查阅异常时的error.log日志发现以下错误:

1
2
3
4
5
[ pid=3413 thr=140583025772288 file=ext/nginx/HelperAgent.cpp:963 time=2014-09-30 11:05:20.925 ]: Uncaught exception in PassengerServer client thread:
exception: Cannot accept new connection: Too many open files (24)
backtrace:
in 'Passenger::FileDescriptor Client::acceptConnection()' (HelperAgent.cpp:429)
in 'void Client::threadMain()' (HelperAgent.cpp:952)

根据日志错误信息,可以确定是PassengerHelperAgent进程文件描述符达到了上限。

找到ext/nginx/HelperAgent.cpp文件的963行:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
void threadMain() {
TRACE_POINT();
try {
while (true) {
UPDATE_TRACE_POINT();
inactivityTimer.start();
FileDescriptor fd(acceptConnection());
inactivityTimer.stop();
handleRequest(fd);
}
} catch (const boost::thread_interrupted &) {
P_TRACE(2, "Client thread " << this << " interrupted.");
} catch (const tracable_exception &e) {
P_ERROR("Uncaught exception in PassengerServer client thread:\n"
<< " exception: " << e.what() << "\n"
<< " backtrace:\n" << e.backtrace());
abort();
}
}

通过上下文可以确定是调用acceptConnection()出错,查看acceptConnection()代码,确定是由该函数抛出的异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
FileDescriptor acceptConnection() {
TRACE_POINT();
struct sockaddr_un addr;
socklen_t addrlen = sizeof(addr);
int fd = syscalls::accept(serverSocket,
(struct sockaddr *) &addr,
&addrlen);
if (fd == -1) {
throw SystemException("Cannot accept new connection", errno);
} else {
return FileDescriptor(fd);
}
}

syscalls::accept是对系统调用accept的简单封装。

1
2
3
4
5
6
7
8
syscalls::accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen) {
int ret;
CHECK_INTERRUPTION(
ret == -1,
ret = ::accept(sockfd, addr, addrlen)
);
return ret;
}

错误原因就是accept由于进程文件描述符达到上限而出错返回了。

接下来追查为什么进程文件描述符数会达到上限。

首先看一下PassengerHelperAgent的整体代码逻辑:

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
int
main(int argc, char *argv[]) {
TRACE_POINT();
VariantMap options = initializeAgent(argc, argv, "PassengerHelperAgent");
pid_t webServerPid = options.getPid("web_server_pid");
string tempDir = options.get("temp_dir");
bool userSwitching = options.getBool("user_switching");
string defaultUser = options.get("default_user");
string defaultGroup = options.get("default_group");
string passengerRoot = options.get("passenger_root");
string rubyCommand = options.get("ruby");
unsigned int generationNumber = options.getInt("generation_number");
unsigned int maxPoolSize = options.getInt("max_pool_size");
unsigned int maxInstancesPerApp = options.getInt("max_instances_per_app");
unsigned int poolIdleTime = options.getInt("pool_idle_time");

try {
UPDATE_TRACE_POINT();
Server server(FEEDBACK_FD, webServerPid, tempDir,
userSwitching, defaultUser, defaultGroup,
passengerRoot, rubyCommand, generationNumber,
maxPoolSize, maxInstancesPerApp, poolIdleTime,
options);
P_DEBUG("Passenger helper agent started on PID " << getpid());

UPDATE_TRACE_POINT();
server.mainLoop();
} catch (const tracable_exception &e) {
P_ERROR(e.what() << "\n" << e.backtrace());
return 1;
} catch (const std::exception &e) {
P_ERROR(e.what());
return 1;
}

P_TRACE(2, "Helper agent exited.");
return 0;
}

首先创建一个Server对象,然后调用Server对象的mainLoop成员函数。Server对象构造函数会调用成员函数startListening。

1
2
3
4
5
6
7
8
9
10
11
12
void startListening() {
this_thread::disable_syscall_interruption dsi;
requestSocket = createUnixServer(getRequestSocketFilename().c_str());

int ret;
do {
ret = chmod(getRequestSocketFilename().c_str(), S_ISVTX |
S_IRUSR | S_IWUSR | S_IXUSR |
S_IRGRP | S_IWGRP | S_IXGRP |
S_IROTH | S_IWOTH | S_IXOTH);
} while (ret == -1 && errno == EINTR);
}

createUnixServer函数会创建一个socket文件,然后监听这个文件。NGINX收到请求后,会由Passenger模块转发请求到该socket文件。
mainLoop会调用成员函数startClientHandlerThreads,它会创建numberOfThreads个Client对象。

1
2
3
4
5
6
7
8
void startClientHandlerThreads() {
for (unsigned int i = 0; i < numberOfThreads; i++) {
ClientPtr client(new Client(i + 1, pool, requestSocketPassword,
defaultUser, defaultGroup, requestSocket,
analyticsLogger));
clients.insert(client);
}
}

Client对象构造函数会启动一个线程执行threadMain。threadMain就是我们上面出错的函数。每个线程等待接收通过socket文件发来的请求,接收请求后调用handleRequest进行处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Client(unsigned int number, ApplicationPool::Ptr pool,
const string &password, const string &defaultUser,
const string &defaultGroup, int serverSocket,
const AnalyticsLoggerPtr &logger)
: inactivityTimer(false)
{
this->number = number;
this->pool = pool;
this->password = password;
this->defaultUser = defaultUser;
this->defaultGroup = defaultGroup;
this->serverSocket = serverSocket;
this->analyticsLogger = logger;

sbmh_init(&statusFinder.ctx, NULL, NULL, 0);
sbmh_init(&transferEncodingFinder.ctx, NULL, NULL, 0);

thr = new oxt::thread(
boost::bind(&Client::threadMain, this),
"Client thread " + toString(number),
CLIENT_THREAD_STACK_SIZE
);
}

问题出在线程调用accept等待接收请求时。我们所创建的线程数量numberOfThreads是在Server对象被创建时指定的。

1
numberOfThreads     = maxPoolSize * 4;

而maxPoolSize由passenger_max_pool_size配置项指定,我们指定的是256。256 × 4 = 1024,开机启动时PassengerHelperAgent进程的文件描述符上限就是1024。这个数字值得怀疑。因而我将配置修改为128,果然正常了。

1
passenger_max_pool_size 256;

可以确定每个线程中占用了文件描述符。然而从代码中并没有找到打开文件相关的逻辑。当accept成功返回时,会返回一个新的文件描述符。开始怀疑accept在还没有接收到请求时就预先占用了一个文件描述符。通过一个简单程序来验证。

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
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/un.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <pthread.h>
#include <assert.h>

void *start_accept(void *arg) {
int fd = (int) arg;
int newfd;
struct sockaddr_un addr;
socklen_t addrlen;

newfd = accept(fd, (struct sockaddr *)&addr, &addrlen);

if (newfd < 0) {
fprintf(stderr, "accept failed: %u: %s\n",
pthread_self(), strerror(errno));
}
}

int main() {
int i, fd;
struct sockaddr_un addr;
socklen_t addrlen = sizeof(addr);
pthread_t pt;

assert((fd = socket(PF_LOCAL, SOCK_STREAM, 0)) > 0);

for (i = 0; i < 1020; i++) {
assert(open("emfile.c", O_RDONLY) > 0);
}

addr.sun_family = AF_LOCAL;
strncpy(addr.sun_path, "emfile.socket", sizeof("emfile.socket") - 1);
addr.sun_path[sizeof("emfile.socket") - 1] = '\0';

assert(bind(fd, (const struct sockaddr *)&addr, sizeof(addr)) == 0);

assert(listen(fd, 512) == 0);

pthread_create(&pt, NULL, start_accept, (void *) fd);

pthread_join(pt, NULL);

exit(0);
}

使用ulimit将shell文件打开数上限修改为1024:

1
$ ulimit -n 1024

编译验证程序,并执行

1
2
$ gcc emfile.c -lpthread
$ ./a.out

得到结果:

1
accept failed: 591611648: Too many open files

确实如些,那再来看一下accept的实现。accept系统调用的内核实现是sys_accept,而sys_accept是对sys_accept4的简单封装。

1
2
3
4
5
SYSCALL_DEFINE3(accept, int, fd, struct sockaddr __user *, upeer_sockaddr,
int __user *, upeer_addrlen)
{
return sys_accept4(fd, upeer_sockaddr, upeer_addrlen, 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
SYSCALL_DEFINE4(accept4, int, fd, struct sockaddr __user *, upeer_sockaddr,
int __user *, upeer_addrlen, int, flags)
{
...

sock = sockfd_lookup_light(fd, &err, &fput_needed);
if (!sock)
goto out;

err = -ENFILE;
if (!(newsock = sock_alloc()))
goto out_put;

newsock->type = sock->type;
newsock->ops = sock->ops;

...

newfd = sock_alloc_file(newsock, &newfile, flags);

...

err = sock->ops->accept(sock, newsock, sock->file->f_flags);
if (err < 0)
goto out_fd;

...

fd_install(newfd, newfile);
err = newfd;

out_put:
fput_light(sock->file, fput_needed);
out:
return err;
out_fd:
fput(newfile);
put_unused_fd(newfd);
goto out_put;
}

sys_accept4中在调用sock->ops->accept去接收网络请求前就调用sock_alloc_file来分配文件描述符。再来看sock_alloc_file这个函数:

1
2
3
4
5
6
7
8
9
10
11
12
static int sock_alloc_file(struct socket *sock, struct file **f, int flags)
{
struct qstr name = { .name = "" };
struct path path;
struct file *file;
int fd;

fd = get_unused_fd_flags(flags);
if (unlikely(fd < 0))
return fd;
...
}

sock_alloc_file会调用get_unused_fd_flags,这是一个宏,实际会调用函数alloc_fd, 而alloc_fd又会调用函数expand_files:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
int expand_files(struct files_struct *files, int nr)
{
struct fdtable *fdt;

fdt = files_fdtable(files);

/*
* N.B. For clone tasks sharing a files structure, this test
* will limit the total number of files that can be opened.
*/
if (nr >= current->signal->rlim[RLIMIT_NOFILE].rlim_cur)
return -EMFILE;

/* Do we need to expand? */
if (nr < fdt->max_fds)
return 0;

/* Can we expand? */
if (nr >= sysctl_nr_open)
return -EMFILE;

/* All good, so we try */
return expand_fdtable(files, nr);
}

可以看到expand_files进行文件描述符限制的检查,当超过限制时返回”EMFILE”。”EMFILE”错误的提示就是”Too many open files”。

1
#define EMFILE      24  /* Too many open files */

结合上面的测试程序,使用一个systemtap脚本可以捕获到上述调用路径。

1
2
3
4
5
6
7
8
9
10
11
12
probe  kernel.function("expand_files").return {
if ($return == -24) {
println("expand_files");
printf("%d\n", $nr);
print_backtrace();
exit();
}
}

probe begin {
println("start\n");
}

执行stap:

1
$sudo stap emfile.stap

捕获结果为:

1
2
3
4
5
6
7
8
9
10
start

expand_files
1024
Returning from: 0xffffffff811931d0 : expand_files+0x0/0x220 [kernel]
Returning to : 0xffffffff81193443 : alloc_fd+0x53/0x160 [kernel]
0xffffffff814187f3 : sock_alloc_file+0x43/0x150 [kernel]
0xffffffff8141b3dd : sys_accept4+0x11d/0x2b0 [kernel]
0xffffffff8141b580 : sys_accept+0x10/0x20 [kernel]
0xffffffff8100b0f2 : system_call_fastpath+0x16/0x1b [kernel]

最终确定异常原因:

开机自启时,进程的打开文件数限制为1024,而创建1024个线程执行accept()时。每个accept会占用一个文件描述符, 达到了进程的文件描述符上限而异常。而从shell启动时,我们shell进程的文件描述符限制是32768,因而不会出现问题。

解决方法:
创建一个启动脚本,在执行/usr/local/nginx/sbin/nginx前执行ulimit修改文件描述符限制。

1
2
ulimit -SHn 65535
/usr/local/nginx/sbin/nginx &

注意:

  • /etc/security/limits.conf中的设置只针对登录动作发生时才生效,因而对于开机自动启动进程这种情况,这种修改该文件的方式不生效。
  • Passenger版本为3.0.11
  • kernel版本为CentOS 6.2内核,kernel-2.6.32-220.4.2.el6

NGINX实现了一套变量机制,使得配置动态内容非常灵活。比如我们可以使用“proxy_cache_key”指令根据我们的需求灵活地设置Cache的KEY。变量机制也为各模块合作提供了一个桥梁,使得模块间配合完成功能更加方便。比如”proxy_pass”指令支持将回源地址设置为变量,我们可以在另一个模块中依据条件对该变量赋值,从而非常简便地完成基于各种策略选择上游地址的功能。

变量机制相关的内部数据结构主要有两种, ngx_http_variable_t和ngx_http_variable_value_t,分别代表变量本身和变量值。

ngx_http_variable_t结构如下:

1
2
3
4
5
6
7
8
struct ngx_http_variable_s {
ngx_str_t name; /* must be first to build the hash */
ngx_http_set_variable_pt set_handler;
ngx_http_get_variable_pt get_handler;
uintptr_t data;
ngx_uint_t flags;
ngx_uint_t index;
};

各成员意义如下:

  • name: 变量名称
  • set_handler: 赋值函数,主要用于”set”指令,处理请求时执行set指令时调用
  • get_handler: 取值函数,当读取该变量时调用该函数得到变量值
  • data: 传递给set_handler和get_handler的参数
  • flags: 变量属性标志
  • index: 变量在cmcf->variables数组中的索引

其中flags取值及意义如下:

  • NGX_HTTP_VAR_CHANGEABLE: 变量被添加时如果已有同名变量,则返回该变量,否则会报错认为变量名冲突。
  • NGX_HTTP_VAR_NOCACHEABLE: 变量的值不应该被缓存。变量被取值后,变量值的no_cacheable被置为1。
  • NGX_HTTP_VAR_INDEXED:表示变量被索引,存储在cmcf->variables数组,这样的变量可以通过索引值直接找到。
  • NGX_HTTP_VAR_NOHASH: 不会将该变量存储在cmcf->variables_hash哈希表。

ngx_http_variable_value_t结构如下:

1
typedef ngx_variable_value_t  ngx_http_variable_value_t;
1
2
3
4
5
6
7
8
9
10
typedef struct {
unsigned len:28;

unsigned valid:1;
unsigned no_cacheable:1;
unsigned not_found:1;
unsigned escape:1;

u_char *data;
} ngx_variable_value_t;

各成员意义如下:

  • len: 变量值数据长度
  • valid: 该变量值是否可用
  • no_cacheable: 该变量值是否不能缓存
  • not_found: 对应变量不存在
  • escape: 变量值内容中的特殊字符是否进行了转义
  • data:变量值的数据

因为NGINX所有变量存储在ngx_http_core_module的main级结构ngx_http_core_main_conf_t中(以下简写cmcf),所以变量的作用范围是整个http{}配置。在某个server{}中添加的变量,在另一个server{}同样可以使用。

1
2
3
4
5
6
7
8
typedef struct {
......
ngx_hash_t variables_hash;
ngx_array_t variables; /* ngx_http_variable_t */
......
ngx_hash_keys_arrays_t *variables_keys;
......
} ngx_http_core_main_conf_t;

NGINX变量有以下3种类型:

  • 模块内置变量
  • 根据配置动态添加的变量
  • 内置规则变量

模块内置变量主要在ngx_http_module_t的preconfiguration阶段中添加。如upstream模块的preconfiguration回调为ngx_http_upstream_add_variables().

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
static ngx_int_t
ngx_http_upstream_add_variables(ngx_conf_t *cf)
{
ngx_http_variable_t *var, *v;

for (v = ngx_http_upstream_vars; v->name.len; v++) {
var = ngx_http_add_variable(cf, &v->name, v->flags);
if (var == NULL) {
return NGX_ERROR;
}

var->get_handler = v->get_handler;
var->data = v->data;
}

return NGX_OK;
}

根据配置动态添加的变量一般当解析相应指令时,在指令的解析函数中添加。比如rewrite模块中的set指令可以添加用户定义名称的变量。

1
2
3
4
v = ngx_http_add_variable(cf, &value[1], NGX_HTTP_VAR_CHANGEABLE);
if (v == NULL) {
return NGX_CONF_ERROR;
}

内置规则变量不需要添加,而是按特定规则来解析。如”http_”, “upstream_http_”, “arg_”, “cookie_”等等一系列变量。
前两种方式都是调用ngx_http_add_variable()来添加变量。ngx_http_add_variable()向cmcf->variable_keys数组中添加变量,并将该变量结构返回。如果指定了NGX_HTTP_VAR_CHANGEABLE标志,那么当检查到同名的变量时,则直接返回该变量。否则报错返回NULL。
如果某一指令需要用到一个变量,则一般在解析该指令配置时会调用ngx_http_get_variable_index(),并将该索引值保存,当处理请求时直接通过索引找到变量。如geo模块中geo指令的解析函数。

1
2
3
4
geo->index = ngx_http_get_variable_index(cf, &name);
if (geo->index == NGX_ERROR) {
return NGX_CONF_ERROR;
}

ngx_http_get_variable_index()会从cmcf->variables数组中查找变量,查找到则返回该变量的索引。否则在cmcf->variables添加该变量并返回数组索引。当解析完HTTP{}配置后,NGINX会将cmcf->variables_keys中的变量组织到cmcf->variables_hash这个HASH表中。如果变量指令了NGX_HTTP_VAR_NOHASH标志,则该变量不会被添加到cmcf->variables_hash中。如果一个变量既没有添加到cmcf->variables中,也没有添加到cmcf->variables_hash中,那么这个变量就不能被找到,因而会被认为不存在。

NGINX开始处理请求时会在请求结构体ngx_http_request_中创建被索引的变量值的缓存空间。

1
2
r->variables = ngx_pcalloc(r->pool, cmcf->variables.nelts
* sizeof(ngx_http_variable_value_t));

对于被索引的变量,可以使用ngx_http_get_indexed_variable()或者ngx_http_get_flushed_variable()来求值。这样省去了查找哈希表的消耗。ngx_http_get_indexed_variable()首先检查r->variables[index]变量缓存是否可用。可用则直接返回,否则调用v->get_handler对变量求值,并将结果存储在r->variables[index]中。

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
ngx_http_variable_value_t *
ngx_http_get_indexed_variable(ngx_http_request_t *r, ngx_uint_t index)
{
ngx_http_variable_t *v;
ngx_http_core_main_conf_t *cmcf;

cmcf = ngx_http_get_module_main_conf(r, ngx_http_core_module);

if (cmcf->variables.nelts <= index) {
ngx_log_error(NGX_LOG_ALERT, r->connection->log, 0,
"unknown variable index: %d", index);
return NULL;
}

if (r->variables[index].not_found || r->variables[index].valid) {
return &r->variables[index];
}

v = cmcf->variables.elts;

if (v[index].get_handler(r, &r->variables[index], v[index].data)
== NGX_OK)
{
if (v[index].flags & NGX_HTTP_VAR_NOCACHEABLE) {
r->variables[index].no_cacheable = 1;
}

return &r->variables[index];
}

r->variables[index].valid = 0;
r->variables[index].not_found = 1;

return NULL;
}

ngx_http_get_flushed_variable()还会对变量值的no_cacheable标志进行检查。如果为0,表示变量值可以cache, 则直接返回已缓存的变量值。否则,将变量值置为不可用,调用ngx_http_get_indexed_variable()对变量求值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
ngx_http_variable_value_t *
ngx_http_get_flushed_variable(ngx_http_request_t *r, ngx_uint_t index)
{
ngx_http_variable_value_t *v;

v = &r->variables[index];

if (v->valid || v->not_found) {
if (!v->no_cacheable) {
return v;
}

v->valid = 0;
v->not_found = 0;
}

return ngx_http_get_indexed_variable(r, index);
}

没有索引的变量,可以调用ngx_http_get_variable()完成取值。它会查找cmcf->variables_hash哈希表,找到变量,从相应变量缓存中取值或调用变量的get_handler。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
cmcf = ngx_http_get_module_main_conf(r, ngx_http_core_module);

v = ngx_hash_find(&cmcf->variables_hash, key, name->data, name->len);

if (v) {

if (v->flags & NGX_HTTP_VAR_INDEXED) {
return ngx_http_get_flushed_variable(r, v->index);

} else {

vv = ngx_palloc(r->pool, sizeof(ngx_http_variable_value_t));

if (vv && v->get_handler(r, vv, v->data) == NGX_OK) {
return vv;
}

return NULL;
}
}

NGINX中rewrite模块实现了set指令,可以给变量赋值。如果另一模块的变量也可以使用set来赋值,则多模块配合完成功能会更加灵活。
set指令的解析函数ngx_http_rewrite_set()代码如下:

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
static char *
ngx_http_rewrite_set(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
ngx_http_rewrite_loc_conf_t *lcf = conf;

ngx_int_t index;
ngx_str_t *value;
ngx_http_variable_t *v;
ngx_http_script_var_code_t *vcode;
ngx_http_script_var_handler_code_t *vhcode;

value = cf->args->elts;

if (value[1].data[0] != '$') {
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
"invalid variable name "%V"", &value[1]);
return NGX_CONF_ERROR;
}

value[1].len--;
value[1].data++;

v = ngx_http_add_variable(cf, &value[1], NGX_HTTP_VAR_CHANGEABLE);
if (v == NULL) {
return NGX_CONF_ERROR;
}

index = ngx_http_get_variable_index(cf, &value[1]);
if (index == NGX_ERROR) {
return NGX_CONF_ERROR;
}

if (v->get_handler == NULL
&& ngx_strncasecmp(value[1].data, (u_char *) "http_", 5) != 0
&& ngx_strncasecmp(value[1].data, (u_char *) "sent_http_", 10) != 0
&& ngx_strncasecmp(value[1].data, (u_char *) "upstream_http_", 14) != 0)
{
v->get_handler = ngx_http_rewrite_var;
v->data = index;
}

if (ngx_http_rewrite_value(cf, lcf, &value[2]) != NGX_CONF_OK) {
return NGX_CONF_ERROR;
}

if (v->set_handler) {
vhcode = ngx_http_script_start_code(cf->pool, &lcf->codes,
sizeof(ngx_http_script_var_handler_code_t));
if (vhcode == NULL) {
return NGX_CONF_ERROR;
}

vhcode->code = ngx_http_script_var_set_handler_code;
vhcode->handler = v->set_handler;
vhcode->data = v->data;

return NGX_CONF_OK;
}

vcode = ngx_http_script_start_code(cf->pool, &lcf->codes,
sizeof(ngx_http_script_var_code_t));
if (vcode == NULL) {
return NGX_CONF_ERROR;
}

vcode->code = ngx_http_script_set_var_code;
vcode->index = (uintptr_t) index;

return NGX_CONF_OK;
}

它向lcf->codes函数引擎添加rewrite阶段需要执行的函数。如果检测到变量的set_handler存在,则添加ngx_http_script_var_set_handler_code函数,它会调用set_handler。而如果没有set_handler, 则添加ngx_http_script_set_var_code函数,它不会调用set_handler。由于内置变量添加一般是在preconfiguration中完成,因而解析set指令时,变量的set_handler存在,可以正常处理。而根据配置动态添加的变量如果解析出现在set指令后,set指令先被解析,此时变量的set_handler为空,此时添加的函数为ngx_http_script_set_var_code,v->set_handler不会得到调用。因此个人感觉添加函数引擎的逻辑应该放到postconfiguration中处理。此时和变量的各成员值都已被正常赋值。因而可以更方便地让根据配置动态添加的变量也可以和set指令轻松结合。

业务要求Cache服务器能够随时增删允许访问的HOST。而每个HOST有单独的配置,这些配置随时都可能更改。如果单纯采用静态配置文件(nginx.conf)的方式,每次修改都要reload NGINX。如果更改很频繁,会造成服务器上存在大量的NGINX进程,导致服务器负载很高。因而我们将需要随时更改的配置存储于一个独立的配置服务器中。请求处理时,先去配置服务器中获取该请求需要使用的配置,再根据这些配置进行相应的处理。因而,我们可以随时更改配置服务器中的相应内容。
其中一个配置就是文件缓存时间。NGINX中设置文件缓存时间有两种方法:

  • 设置proxy_cache_valid指令
  • 在上游响应中的添加”Cache-Control” header和”Expires” header

其中上游响应header的优先级更高。当不想使用上游响应header中所设置的缓存时间时,可以使用以下指令来禁用。

1
proxy_ignore_headers X-Accel-Expires;

这两种方法都无法满足我们根据动态配置来设置缓存时间的需求。因而我给NGINX添加了一个内置变量”cache_time”来支持灵活地设置缓存时间,并且该种方式具有最高的优先级。这样,可以非常方便地在ngx_lua等第三方模块中根据条件设置不同的缓存时间。

在ngx_http_request_t添加一个cache_time成员,在ngx_http_core_variables数组中添加内置变量”cache_time”,”cache_time”在被赋值时会将值存储在r->cache_time中。

1
2
3
4
{ ngx_string("cache_time"), ngx_http_variable_request_set_time,
ngx_http_variable_request_get_time,
offsetof(ngx_http_request_t, cache_time),
NGX_HTTP_VAR_CHANGEABLE|NGX_HTTP_VAR_NOCACHEABLE, 0 },
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
static void
ngx_http_variable_request_set_time(ngx_http_request_t *r,
ngx_http_variable_value_t *v, uintptr_t data)
{
ngx_str_t val;
time_t valid, *vp;

val.len = v->len;
val.data = v->data;

valid = ngx_parse_time(&val, 1);
if (valid == (time_t) NGX_ERROR) {
ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
"invalid time value "%V"", &val);
return;
}

vp = (time_t *) ((char *) r + data);

*vp = valid;

return;
}

因为”cache_time”变量需要比上游响应header具有更高的优先级,因而要在上游header处理之后再处理”cache_time”变量。上游响应的header在ngx_http_upstream_process_headers()中进行处理。因而我在upstream模块中添加了一个hook, 该hook在调用完ngx_http_upstream_process_headers()后,开始处理body前被调用。

1
2
3
4
5
6
7
8
9
10
11
if (ngx_http_upstream_process_headers(r, u) != NGX_OK) {
return;
}

if (u->post_headers) {
rc = u->post_headers(r);
if (rc != NGX_OK) {
ngx_http_upstream_finalize_request(r, u, rc);
return;
}
}

proxy模块在该hook上注册一个函数,这个函数执行时,首先检查上游响应的状态码判断是否需要处理”cache_time”变量。检查通过后,读取”cache_time”变量的值,依据值来进行各种操作。当值为0时,禁用cache.为正值,则将缓存时间修改为该值。当修改cache缓存时间后,将上游响应中的”Cache-Control”和”Expires” header去除,不再发送给下游。

1
u->post_headers = ngx_http_proxy_post_headers;
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
static ngx_int_t
ngx_http_proxy_post_headers(ngx_http_request_t *r)
{
ngx_uint_t i;
ngx_table_elt_t **ph;

if (ngx_http_upstream_check_status(r->upstream->conf->cache_time_valid,
r->upstream->headers_in.status_n)
== NGX_DECLINED)
{
return NGX_OK;
}

if (r->cache_time == (time_t) -1) {
return NGX_OK;
}

if (r->cache_time == (time_t) 0) {
r->upstream->cacheable = 0;
return NGX_OK;
}

r->cache->valid_sec = ngx_time() + r->cache_time;

r->headers_out.expires->hash = 0;

ph = r->headers_out.cache_control.elts;
for (i = 0; i < r->headers_out.cache_control.nelts; i++) {
ph[i]->hash = 0;
}

return NGX_OK;
}

XEP-0114中定义了Jabber组件协议(Jabber Componet Protocol)。XMPP网络外的可信组件可以使用这个协议和XMPP网络内实体进行通信。

组件协议定义了两种模式:

  • accept:外部组件向XMPP服务器发起连接
  • connect:XMPP服务器向外部组件发起连接

其中, accept方式使用比较广泛,ejabberd中只实现了accept方式。

组件协议像XMPP一样,也是基于XML流,使用的XMLNS为jabber:componet:accept或者jabber:component:connect

accept方式的协议流程:

  • 外部组件建立到XMPP服务器的TCP连接,发送流头。

    1
    2
    3
    4
    <stream:stream
    xmlns='jabber:component:accept'
    xmlns:stream='http://etherx.jabber.org/streams'
    to='plays.shakespeare.lit'>
  • XMPP服务器回应,也发送流头,其中必须包括流ID属性:

    1
    2
    3
    4
    5
    <stream:stream
    xmlns:stream='http://etherx.jabber.org/streams'
    xmlns='jabber:component:accept'
    from='plays.shakespeare.lit'
    id='3BF96D32'>
  • 外部组件发送身份验证摘要信息。

    1
    <handshake>aaee83c26aeeafcbabeabfcbcd50df997e0a2a1e</handshake>

    组件协议身份验证不使用SASL,也不使用已废弃的XEP-0078。它使用双方共享密钥计算摘要信息来验证身份。计算方法如下:

    1. 将服务器流头中的流ID属性和共享密钥拼接成字符串
    2. 计算该字符串的SHA1哈希值,并转换成小写16进制字符串
  • XMPP服务器用同样方法计算进行校验。通过后,返回一个空的handshake元素。

    1
    <handshake/>

    至此,外部组件和XMPP服务器就可以交换XMPP消息了。

我们来看ejabberd中组件协议实现,位于ejabberd_service.erl模块中。

ejabberd中的ejabberd_service的默认配置为:

1
2
3
4
5
6
7
8
{8888, ejabberd_service, [
{access, all},
{shaper_rule, fast},
{ip, {127, 0, 0, 1}},
{hosts, ["icq.example.org", "sms.example.org"],
[{password, "secret"}]
}
]},

ejabberd_service是端口8888的处理模块。当有ejabberd接收端口上的TCP连接后,ejabberd_socket:start/4调用处理模块的socket_type/0, 根据返回值进行不同处理。ejabberd_service:socket_type/0返回xml_stream。它的处理流程和ejabberd_c2s模块相同。ejabberd为每个TCP连接分别创建一个receiver进程和一个处理进程(这里是ejabberd_service进程)。receiver进程接收消息并解析,然后发送相应的消息给处理进程。具体不再详述,请参考:ejabberd消息处理流程分析

service进程为gen_fsm进程,初始状态为wait_for_stream。receiver进程接收到XML流头后发送xmlstreamstart消息给service进程。service进程调用wait_for_stream函数进行处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
wait_for_stream({xmlstreamstart, _Name, Attrs}, StateData) ->
case xml:get_attr_s("xmlns", Attrs) of
"jabber:component:accept" ->
%% Note: XEP-0114 requires to check that destination is a Jabber
%% component served by this Jabber server.
%% However several transports don't respect that,
%% so ejabberd doesn't check 'to' attribute (EJAB-717)
To = xml:get_attr_s("to", Attrs),
Header = io_lib:format(?STREAM_HEADER,
[StateData#state.streamid, xml:crypt(To)]),
send_text(StateData, Header),
{next_state, wait_for_handshake, StateData};
_ ->
send_text(StateData, ?INVALID_HEADER_ERR),
{stop, normal, StateData}
end;

wait_for_stream检测到XML流头中XMLNS为”jabber:component:accept”后,向组件发送流头,状态变更为wait_for_handshake。

receiver进程收到handshake消息后,发送xmlstreamelement消息给service进程,service调用wait_for_handshake处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
wait_for_handshake({xmlstreamelement, El}, StateData) ->
{xmlelement, Name, _Attrs, Els} = El,
case {Name, xml:get_cdata(Els)} of
{"handshake", Digest} ->
case sha:sha(StateData#state.streamid ++
StateData#state.password) of
Digest ->
send_text(StateData, "<handshake/>"),
lists:foreach(
fun(H) ->
ejabberd_router:register_route(H),
?INFO_MSG("Route registered for service ~p~n", [H])
end, StateData#state.hosts),
{next_state, stream_established, StateData};
_ ->
send_text(StateData, ?INVALID_HANDSHAKE_ERR),
{stop, normal, StateData}
end;
_ ->
{next_state, wait_for_handshake, StateData}
end;

wait_for_handshake使用XML流ID和密码计算身份验证摘要,和组件所发的摘要信息进行对比判断是否通过。检查通过后,发送空的handshake元素。然后调用ejabberd_router:register_route/1依次注册配置的所有service域名。这样,XMPP实体发往这些域名的消息都将被ejabberd_router路由给该service进程。service进程状态变更为stream_established。

至此,外部组件就可以和XMPP服务器交换XMPP消息了。

组件向XMPP服务器发送消息后,receiver进程解析后向service进程发送xmlstreamelement消息,service进程调用stream_established处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
stream_established({xmlstreamelement, El}, StateData) ->
NewEl = jlib:remove_attr("xmlns", El),
{xmlelement, Name, Attrs, _Els} = NewEl,
From = xml:get_attr_s("from", Attrs),
...
To = xml:get_attr_s("to", Attrs),
ToJID = case To of
"" -> error;
_ -> jlib:string_to_jid(To)
end,
if
((Name == "iq") or
(Name == "message") or
(Name == "presence")) and
(ToJID /= error) and (FromJID /= error) ->
ejabberd_router:route(FromJID, ToJID, NewEl);
true ->
Err = jlib:make_error_reply(NewEl, ?ERR_BAD_REQUEST),
send_element(StateData, Err),
error
end,
{next_state, stream_established, StateData};

stream_established进行一系列检查后,调用ejabberd_router:route转发消息。

XMPP实体发送给service域名的消息会由ejabberd_router以route消息的格式发给service进程。service进程调用handle_info处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
handle_info({route, From, To, Packet}, StateName, StateData) ->
case acl:match_rule(global, StateData#state.access, From) of
allow ->
{xmlelement, Name, Attrs, Els} = Packet,
Attrs2 = jlib:replace_from_to_attrs(jlib:jid_to_string(From),
jlib:jid_to_string(To),
Attrs),
Text = xml:element_to_binary({xmlelement, Name, Attrs2, Els}),
send_text(StateData, Text);
deny ->
Err = jlib:make_error_reply(Packet, ?ERR_NOT_ALLOWED),
ejabberd_router:route_error(To, From, Err, Packet)
end,
{next_state, StateName, StateData};

handle_info首先进行ACL检查,通过后,修改From和To属性,将消息发送给组件。

使用telnet演示简单登录过程:

1
2
3
4
5
6
7
8
9
10
[root@flygoast flygoast]# telnet 127.0.0.1 8888
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
<stream:stream
xmlns='jabber:component:accept'
xmlns:stream='http://etherx.jabber.org/streams'
to='sms.example.com'>
<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:component:accept' id='2744762983' from='sms.example.com'><handshake>cffc7fab4feae018a325ea834d2dca8c3b707a51</handshake>
<handshake/>

身份校验信息计算:

1
2
[root@flygoast flygoast]# echo -n "2744762983secret" | sha1sum
cffc7fab4feae018a325ea834d2dca8c3b707a51 -

根据XEP-0199, XMPP客户端和服务器都可以在XML流上发送应用层PING请求。因为XMPP依赖底层的TCP连接,有可能TCP连接意外中断,而上层的XMPP并不知晓,从而影响消息传递。通过发送应用层PING请求可以来确认对端的连接可用性。

以服务器发给客户端为例,协议如下:

发送的PING请求:

1
2
3
<iq from='capulet.lit' to='juliet@capulet.lit/balcony' id='s2c1' type='get'>
<ping xmlns='urn:xmpp:ping'/>
</iq>

如果对端支持PING请求,则返回对应的”PONG”回应。

1
<iq from='juliet@capulet.lit/balcony' to='capulet.lit' id='s2c1' type='result'/>

如果对端不支持则返回错误。

1
2
3
4
5
6
<iq from='juliet@capulet.lit/balcony' to='capulet.lit' id='s2c1' type='error'>
<ping xmlns='urn:xmpp:ping'/>
<error type='cancel'>
<service-unavailable xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/>
</error>
</iq>

ejabberd中PING功能实现位于mod_ping.erl。它主要支持3个配置:

  • send_pings: true|false

如果这个选项设置为true, 当客户端在给定时间间隔内没有活动,则向客户端发送一个ping请求。

  • ping_interval: Seconds

设置上述send_pings选项中客户端没有活动的时间间隔。

  • timeout_action: none|kill

表示当PING请求发出32秒后,ejabberd依然没有收到PING响应,服务端如何处理。none表示什么也不做,kill表示关闭客户端连接。

当ejabberd启动时会调用mod_ping:start/2。

1
2
3
4
5
start(Host, Opts) ->
Proc = gen_mod:get_module_proc(Host, ?MODULE),
PingSpec = {Proc, {?MODULE, start_link, [Host, Opts]},
transient, 2000, worker, [?MODULE]},
supervisor:start_child(?SUPERVISOR, PingSpec).

start函数调用supervisor:start_child/2为每个支持的host创建一个负责该host的worker进程。

进程树模型如下:

token data
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
+------------+
|ejabberd_sup|
+-----+------+
|
| +------------------+
+--->|Other processes...|
| +------------------+
|
| +------------------------+
+--->|ping(im.just4coding.com)|
| +------------------------+
|
| +------------------------+
+--->|ping(localhost) |
| +------------------------+
|
| +------------------------+
+--->|ping(Other host) |
+------------------------+

每个worker是一个gen_server进程,进程调用init函数进行初始化。

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
init([Host, Opts]) ->
SendPings = gen_mod:get_opt(send_pings, Opts, ?DEFAULT_SEND_PINGS),
PingInterval = gen_mod:get_opt(ping_interval, Opts, ?DEFAULT_PING_INTERVAL),
TimeoutAction = gen_mod:get_opt(timeout_action, Opts, none),
IQDisc = gen_mod:get_opt(iqdisc, Opts, no_queue),
mod_disco:register_feature(Host, ?NS_PING),
gen_iq_handler:add_iq_handler(ejabberd_sm, Host, ?NS_PING,
?MODULE, iq_ping, IQDisc),
gen_iq_handler:add_iq_handler(ejabberd_local, Host, ?NS_PING,
?MODULE, iq_ping, IQDisc),
case SendPings of
true ->
%% Ping requests are sent to all entities, whether they
%% announce 'urn:xmpp:ping' in their caps or not
ejabberd_hooks:add(sm_register_connection_hook, Host,
?MODULE, user_online, 100),
ejabberd_hooks:add(sm_remove_connection_hook, Host,
?MODULE, user_offline, 100),
ejabberd_hooks:add(user_send_packet, Host,
?MODULE, user_send, 100);
_ ->
ok
end,
{ok, #state{host = Host,
send_pings = SendPings,
ping_interval = PingInterval,
timeout_action = TimeoutAction,
timers = ?DICT:new()}}.
  • 首先获取相关配置

  • 接着调用mod_disco:register_feature注册PING功能的XMLNS。这样当客户端请求”Service Discovery”信息时,ejabberd返回的特征中会包括”urn:xmpp:ping”。

ServiceDiscovery请求:

1
2
3
4
5
6
<iq type='get'
from='juliet@capulet.lit/balcony'
to='capulet.lit'
id='disco1'>
<query xmlns='http://jabber.org/protocol/disco#info'/>
</iq>

ServiceDiscovery响应:

1
2
3
4
5
6
7
8
9
10
<iq type='result'
from='capulet.lit'
to='juliet@capulet.lit/balcony'
id='disco1'>
<query xmlns='http://jabber.org/protocol/disco#info'>
...
<feature var='urn:xmpp:ping'/>
...
</query>
</iq>

ServiceDiscovery相关信息参考XEP-0030

  • 接下来,注册IQ处理器,令XMLNS为”urn:xmpp:ping”的IQ请求由函数iq_ping处理。iq_ping简单地返回相应响应或者错误。

    1
    2
    3
    4
    5
    6
    7
    iq_ping(_From, _To, #iq{type = Type, sub_el = SubEl} = IQ) ->
    case {Type, SubEl} of
    {get, {xmlelement, "ping", _, _}} ->
    IQ#iq{type = result, sub_el = []};
    _ ->
    IQ#iq{type = error, sub_el = [SubEl, ?ERR_FEATURE_NOT_IMPLEMENTED]}
    end.

    如果send_pings配置为true, mod_ping在ejabberd中注册n以下3个hook函数:

  • sm_register_connection_hook: 它在客户端完成登录验证,建立session信息时调用。

1
2
3
4
5
6
7
8
open_session(SID, User, Server, Resource, Info) ->
set_session(SID, User, Server, Resource, undefined, Info),
mnesia:dirty_update_counter(session_counter,
jlib:nameprep(Server), 1),
check_for_sessions_to_replace(User, Server, Resource),
JID = jlib:make_jid(User, Server, Resource),
ejabberd_hooks:run(sm_register_connection_hook, JID#jid.lserver,
[SID, JID, Info]).
  • sm_remove_connection_hook: 在用户退出,关闭session时调用。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    close_session(SID, User, Server, Resource) ->
    Info = case mnesia:dirty_read({session, SID}) of
    [] -> [];
    [#session{info=I}] -> I
    end,
    F = fun() ->
    mnesia:delete({session, SID}),
    mnesia:dirty_update_counter(session_counter,
    jlib:nameprep(Server), -1)
    end,
    mnesia:sync_dirty(F),
    JID = jlib:make_jid(User, Server, Resource),
    ejabberd_hooks:run(sm_remove_connection_hook, JID#jid.lserver,
    [SID, JID, Info]).
  • user_send_packet: 在C2S进程收到客户端发送的消息时被调用。

sm_register_connection_hook的hook函数user_onlineuser_send_packet的hook函数user_send都会调用start_ping函数。

1
2
3
start_ping(Host, JID) ->
Proc = gen_mod:get_module_proc(Host, ?MODULE),
gen_server:cast(Proc, {start_ping, JID}).

start_ping向该HOST的worker进程发送一个{start_ping, JID}消息。worker进程调用handle_cast进行处理:

1
2
3
handle_cast({start_ping, JID}, State) ->
Timers = add_timer(JID, State#state.ping_interval, State#state.timers),
{noreply, State#state{timers = Timers}};

handle_cast调用add_timer为该客户端创建一个timer。

1
2
3
4
5
6
7
8
9
10
11
add_timer(JID, Interval, Timers) ->
LJID = jlib:jid_tolower(JID),
NewTimers = case ?DICT:find(LJID, Timers) of
{ok, OldTRef} ->
cancel_timer(OldTRef),
?DICT:erase(LJID, Timers);
_ ->
Timers
end,
TRef = erlang:start_timer(Interval * 1000, self(), {ping, JID}),
?DICT:store(LJID, TRef, NewTimers).

由于用户每次发送消息时都会调用add_timer函数,因而add_timer中需要检查之前是否已经存在timer。如果存在,则先取消旧的timer, 再创建新的Timer。

当timer超时后,即客户若干时间内没有活动,进程收到{ping, JID}消息,此时ejabberd应向客户端发送PING消息。进程调用handle_info处理。

1
2
3
4
5
6
7
8
9
10
11
handle_info({timeout, _TRef, {ping, JID}}, State) ->
IQ = #iq{type = get,
sub_el = [{xmlelement, "ping", [{"xmlns", ?NS_PING}], []}]},
Pid = self(),
F = fun(Response) ->
gen_server:cast(Pid, {iq_pong, JID, Response})
end,
From = jlib:make_jid("", State#state.host, ""),
ejabberd_local:route_iq(From, JID, IQ, F),
Timers = add_timer(JID, State#state.ping_interval, State#state.timers),
{noreply, State#state{timers = Timers}};

handle_info创建IQ消息后,设置回调函数F,调用ejabberd_local:route_iq/4消息IQ消息发送给客户端。当收到该IQ消息的响应或者超过32秒依然没有收到客户端的响应,回调函数F将会被调用。如果响应超时,Response为timeout,F将向进程发送{iq_pong, JID, timeout}消息。进程调用handle_cast处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
handle_cast({iq_pong, JID, timeout}, State) ->
Timers = del_timer(JID, State#state.timers),
ejabberd_hooks:run(user_ping_timeout, State#state.host, [JID]),
case State#state.timeout_action of
kill ->
#jid{user = User, server = Server, resource = Resource} = JID,
case ejabberd_sm:get_session_pid(User, Server, Resource) of
Pid when is_pid(Pid) ->
ejabberd_c2s:stop(Pid);
_ ->
ok
end;
_ ->
ok
end,
{noreply, State#state{timers = Timers}};

如果timeout_action设置为kill, 则调用ejabberd_c2s:stop关闭相应的客户端连接。

因为在sm_remove_connection_hook注册了hook函数user_offline, 当用户退出时会调用stop_ping函数,向worker进程发送{stop_ping, JID}消息。

1
2
3
stop_ping(Host, JID) ->
Proc = gen_mod:get_module_proc(Host, ?MODULE),
gen_server:cast(Proc, {stop_ping, JID}).

worker进程调用del_timer函数将该客户端的timer删除。

1
2
3
handle_cast({stop_ping, JID}, State) ->
Timers = del_timer(JID, State#state.timers),
{noreply, State#state{timers = Timers}};
1
2
3
4
5
6
7
8
9
del_timer(JID, Timers) ->
LJID = jlib:jid_tolower(JID),
case ?DICT:find(LJID, Timers) of
{ok, TRef} ->
cancel_timer(TRef),
?DICT:erase(LJID, Timers);
_ ->
Timers
end.

模块及进程停止的逻辑与模块和进程初始化的逻辑相反,本文略过。

NGINX中使用ngx_http_output_filter()向一个请求发送响应。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
ngx_int_t
ngx_http_output_filter(ngx_http_request_t *r, ngx_chain_t *in)
{
ngx_int_t rc;
ngx_connection_t *c;

c = r->connection;

ngx_log_debug2(NGX_LOG_DEBUG_HTTP, c->log, 0,
"http output filter "%V?%V"", &r->uri, &r->args);

rc = ngx_http_top_body_filter(r, in);

if (rc == NGX_ERROR) {
/* NGX_ERROR may be returned by any filter */
c->error = 1;
}

return rc;
}

其中,r是请求结构体,in为以ngx_chain_t结构链接起来的需要发送的内容。ngx_http_output_filter()会调用ngx_http_top_body_filter()。NGINX采用filter机制对响应进行处理。filter分为header filterbody filter。NGINX分别将两种filter各构建成一个链表。全局变量ngx_http_top_header_filter指向header filter链表的头结点,而全局变量ngx_http_top_body_filter指向body filter链表的头结点。每个filter中用一个变量记录下一个filter,依据情况决定是调用下一个filter,还是直接返回。

body filter链表顺序如下图:

token data
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

+--------------------------+
|ngx_http_range_body_filter|
+----------+---------------+
|
v
+----------+---------+
|ngx_http_copy_filter|
+----------+---------+
|
v
+----------+-----------------+
|ngx_http_charset_body_filter|
+----------+-----------------+
|
v
+----------+-------------+
|ngx_http_ssi_body_filter|
+----------+-------------+
|
v
+----------+-------------+
|ngx_http_postpone_filter|
+----------+-------------+
|
v
+----------+--------------+
|ngx_http_gzip_body_filter|
+----------+--------------+
|
v
+----------+-----------------+
|ngx_http_chunked_body_filter|
+----------+-----------------+
|
v
+---------------------+
|ngx_http_write_filter|
+---------------------+

header filter链表顺序如图:

token data
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

+----------------------------+
|ngx_http_not_modified_filter|
+----------+-----------------+
|
v
+----------+------------+
|ngx_http_headers_filter|
+----------+------------+
|
v
+----------+-----------+
|ngx_http_userid_filter|
+----------+-----------+
|
v
+----------+-------------------+
|ngx_http_charset_header_filter|
+----------+-------------------+
|
v
+----------+---------------+
|ngx_http_ssi_header_filter|
+----------+---------------+
|
v
+----------+----------------+
|ngx_http_gzip_header_filter|
+----------+----------------+
|
v
+----------+-----------------+
|ngx_http_range_header_filter|
+----------+-----------------+
|
v
+----------+-------------------+
|ngx_http_chunked_header_filter|
+----------+-------------------+
|
v
+----------+-----------+
|ngx_http_header_filter|
+----------------------+

ngx_http_write_filter()是最后一个被调用的body filter,它进行真正的网络I/O操作,将响应发送给客户端。实际上, ngx_http_header_filter()也是调用ngx_http_write_filter()来发送响应中的headers。

ngx_http_write_filter()的简化流程如下:

  • 检查之前是否有错误发生(c->error被置位)。如果有错误发生,则没有必要再进行网络I/O操作,直接返回NGX_ERROR。
    1
    2
    3
    if (c->error) {
    return NGX_ERROR;
    }
  • 计算之前没有发送完成的内容大小并检查是否存在特殊标志。为了优化性能,当没有必要立即发送响应且响应内容大小没有达到设置的阀值时,NGINX可以暂时推迟发送该部分响应。参看:postpone_output指令。flush标志表示需要立即发送响应。recycled表示该buffer需要循环使用,因而需要立即发送以释放该buffer被重新使用。last标志表示该buffer是响应的最后一部分内容,因而也需要立即发送。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
for (cl = r->out; cl; cl = cl->next) {
ll = &cl->next;

......

size += ngx_buf_size(cl->buf);

if (cl->buf->flush || cl->buf->recycled) {
flush = 1;
}

if (cl->buf->last_buf) {
last = 1;
}
}
  • 计算本次将发送的内容大小,检查是否存在特殊标志,并将内容链接到r->out上。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    for (ln = in; ln; ln = ln->next) {
    cl = ngx_alloc_chain_link(r->pool);
    if (cl == NULL) {
    return NGX_ERROR;
    }

    cl->buf = ln->buf;
    *ll = cl;
    ll = &cl->next;

    ...

    size += ngx_buf_size(cl->buf);

    if (cl->buf->flush || cl->buf->recycled) {
    flush = 1;
    }

    if (cl->buf->last_buf) {
    last = 1;
    }
    }
  • 根据情况决定是需要真正进行网络I/O操作, 还是直接返回。
    1
    2
    3
    if (!last && !flush && in && size < (off_t) clcf->postpone_output) {
    return NGX_OK;
    }
  • 真正进行网络I/O操作,发送内容。
    1
    chain = c->send_chain(c, r->out, limit);
  • 回收发送完成内容的buffer和chain结构, 将没有发送完成的内容存入r->out
    1
    2
    3
    4
    5
    6
    7
    for (cl = r->out; cl && cl != chain; /* void */) {
    ln = cl;
    cl = cl->next;
    ngx_free_chain(r->pool, ln);
    }

    r->out = chain;
  • 根据发送是否完成,返回NGX_OKNGX_AGAIN
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    if (chain) {
    c->buffered |= NGX_HTTP_WRITE_BUFFERED;
    return NGX_AGAIN;
    }

    c->buffered &= ~NGX_HTTP_WRITE_BUFFERED;

    if ((c->buffered & NGX_LOWLEVEL_BUFFERED) && r->postponed == NULL) {
    return NGX_AGAIN;
    }

    return NGX_OK;
    此外,ngx_http_write_filter()中也处理了限速发送的逻辑,本文不详述。

XMPP使用SASL进行登录验证。ejabberd中模仿CyrusSASL库自己实现了SASL协议,API用法与CyrusSASL类似。

ejabberd启动时,ejabberd_app:start/0会调用cyrsasl:start/0。它创建了一个ets表sasl_mechanism,然后调用各个SASL机制模块的start函数。

1
2
3
4
5
6
7
8
9
start() ->
ets:new(sasl_mechanism, [named_table,
public,
{keypos, #sasl_mechanism.mechanism}]),
cyrsasl_plain:start([]),
cyrsasl_digest:start([]),
cyrsasl_scram:start([]),
cyrsasl_anonymous:start([]),
ok.

各个SASL机制模块的start/1函数会调用cyrsasl:register_mechanism/3, 参数分别为SASL机制名称,机制处理模块,机制支持的密码存储类型。register_mechanism会将机制信息添加到sasl_mechanism表中。

1
2
3
start(_Opts) ->
cyrsasl:register_mechanism("PLAIN", ?MODULE, plain),
ok.
1
2
3
4
5
register_mechanism(Mechanism, Module, PasswordType) ->
ets:insert(sasl_mechanism,
#sasl_mechanism{mechanism = Mechanism,
module = Module,
password_type = PasswordType}).

客户端连接到ejabberd后,ejabberd会创建两个进程。一个ejabberd_c2s进程来处理连接状态并向客户端发送数据,另一个ejabberd_receiver进程接收客户端数据。ejabberd_c2s进程实现了gen_fsm行为,有限状态机的初始状态为wait_for_stream。ejabberd_c2s:init/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
init([{SockMod, Socket}, Opts]) ->
...
IP = peerip(SockMod, Socket),
%% Check if IP is blacklisted:
case is_ip_blacklisted(IP) of
true ->
?INFO_MSG("Connection attempt from blacklisted IP: ~s (~w)",
[jlib:ip_to_list(IP), IP]),
{stop, normal};
false ->
...
{ok, wait_for_stream, #state{socket = Socket1,
sockmod = SockMod,
socket_monitor = SocketMonitor,
xml_socket = XMLSocket,
zlib = Zlib,
tls = TLS,
tls_required = StartTLSRequired,
tls_enabled = TLSEnabled,
tls_options = TLSOpts,
streamid = new_id(),
access = Access,
shaper = Shaper,
ip = IP},
?C2S_OPEN_TIMEOUT}
end.

ejabberd_receiver进程解析到XML流头之后会调用gen_fsm:send_event向ejabberd_c2s进程发送消息。

1
2
3
4
5
6
7
8
9
10
11
12
process_data([Element|Els], #state{c2s_pid = C2SPid} = State)
when element(1, Element) == xmlelement;
element(1, Element) == xmlstreamstart;
element(1, Element) == xmlstreamelement;
element(1, Element) == xmlstreamend ->
if
C2SPid == undefined ->
State;
true ->
catch gen_fsm:send_event(C2SPid, element_wrapper(Element)),
process_data(Els, State)
end;

ejabberd_c2s进程接收到{xmlstreamstart, _Name, Attrs}消息后,调用状态函数wait_for_stream/2来处理。wait_for_stream在一系列正确性校验通过之后,回应给客户端XML流头。如果这个XML流之前还没有通过登录验证,则进行登录验证过程。简化的代码逻辑如下:

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
case StateData#state.authenticated of
false ->
SASLState = cyrsasl:server_new(
"jabber", Server, "", [],
fun(U) ->
ejabberd_auth:get_password_with_authmodule(U, Server)
end,
fun(U, P) ->
ejabberd_auth:check_password_with_authmodule(U, Server, P)
end,
fun(U, P, D, DG) ->
ejabberd_auth:check_password_with_authmodule(U, Server, P, D, DG)
end),
Mechs = lists:map(
fun(S) ->
{xmlelement, "mechanism", [], [{xmlcdata, S}]}
end, cyrsasl:listmech(Server)),
...

send_element(StateData,
{xmlelement, "stream:features", [],
TLSFeature ++ CompressFeature ++
[{xmlelement, "mechanisms",
[{"xmlns", ?NS_SASL}],
Mechs}] ++
ejabberd_hooks:run_fold(
c2s_stream_features,
Server,
[], [Server])}),
fsm_next_state(wait_for_feature_request,
StateData#state{
server = Server,
sasl_state = SASLState,
lang = Lang});
_ ->
...
end

首先调用cyrsasl:server_new/7创建一个SASL验证状态, 其中存储了3个用于密码校验的回调函数。

1
2
3
4
5
6
7
8
server_new(Service, ServerFQDN, UserRealm, _SecFlags,
GetPassword, CheckPassword, CheckPasswordDigest) ->
#sasl_state{service = Service,
myname = ServerFQDN,
realm = UserRealm,
get_password = GetPassword,
check_password = CheckPassword,
check_password_digest= CheckPasswordDigest}.

wait_for_stream函数接着调用cyrsasl:listmech/1获取当前域名所支持的SASL验证机制。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
listmech(Host) ->
Mechs = ets:select(sasl_mechanism,
[{#sasl_mechanism{mechanism = '$1',
password_type = '$2',
_ = '_'},
case catch ejabberd_auth:store_type(Host) of
external ->
[{'==', '$2', plain}];
scram ->
[{'/=', '$2', digest}];
{'EXIT',{undef,[{Module,store_type,[]} | _]}} ->
?WARNING_MSG("~p doesn't implement the function store_type/0", [Module]),
[];
_Else ->
[]
end,
['$1']}]),
filter_anonymous(Host, Mechs).

listmech函数调用ejabberd_auth:store_type/1从ejabberd.cfg文件获取密码存储格式(auth_password_format)的配置。从sasl_mechanism表中查询出支持该密码存储格式的SASL机制。wait_for_stream函数将这些机制组织成XMPP协议格式发送回客户端,将当前进程状态改为wait_for_feature_request。比如,发送的机制列表为:

1
2
3
4
5
6
7
8
9
<stream:features>
<mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl">
<mechanism>PLAIN</mechanism>
<mechanism>DIGEST-MD5</mechanism>
<mechanism>SCRAM-SHA-1</mechanism>
</mechanisms>
<c xmlns="http://jabber.org/protocol/caps" node="http://www.process-one.net/en/ejabberd/" ver="yy7di5kE0syuCXOQTXNBTclpNTo=" hash="sha-1"/>
<register xmlns="http://jabber.org/features/iq-register"/>
</stream:features>

客户端从其中选择一个机制并发送给ejabberd服务器。如:

1
<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="PLAIN">AGFhYQAxMjM=</auth>

ejabberd_receiver进程解析完这个XML元素后,发送消息{xmlstreamelement, El}给ejabberd_c2s进程。ejabberd_c2s进程调用wait_for_feature_request函数进行处理。

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
Mech = xml:get_attr_s("mechanism", Attrs),
ClientIn = jlib:decode_base64(xml:get_cdata(Els)),
case cyrsasl:server_start(StateData#state.sasl_state,
Mech,
ClientIn) of
{ok, Props} ->
(StateData#state.sockmod):reset_stream(StateData#state.socket),
send_element(StateData, {xmlelement, "success", [{"xmlns", ?NS_SASL}], []}),
U = xml:get_attr_s(username, Props),
AuthModule = xml:get_attr_s(auth_module, Props),
...
fsm_next_state(wait_for_stream,
StateData#state{
streamid = new_id(),
authenticated = true,
auth_module = AuthModule,
user = U });
{continue, ServerOut, NewSASLState} ->
send_element(StateData,
{xmlelement, "challenge",
[{"xmlns", ?NS_SASL}],
[{xmlcdata,
jlib:encode_base64(ServerOut)}]}),
fsm_next_state(wait_for_sasl_response,
StateData#state{
sasl_state = NewSASLState});
{error, Error, Username} ->
IP = peerip(StateData#state.sockmod, StateData#state.socket),
...
send_element(StateData,
{xmlelement, "failure",
[{"xmlns", ?NS_SASL}],
[{xmlelement, Error, [], []}]}),
{next_state, wait_for_feature_request, StateData, ?C2S_OPEN_TIMEOUT};
{error, Error} ->
send_element(StateData,
{xmlelement, "failure",
[{"xmlns", ?NS_SASL}],
[{xmlelement, Error, [], []}]}),
fsm_next_state(wait_for_feature_request, StateData)
end;

ejabberd_c2s进程判断收到的XML元素是auth请求后,从请求中获取客户端选择的机制mechanism,读取客户端发送的信息,然后调用cyrsasl:server_start/3。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
server_start(State, Mech, ClientIn) ->
case lists:member(Mech, listmech(State#sasl_state.myname)) of
true ->
case ets:lookup(sasl_mechanism, Mech) of
[#sasl_mechanism{module = Module}] ->
{ok, MechState} = Module:mech_new(
State#sasl_state.myname,
State#sasl_state.get_password,
State#sasl_state.check_password,
State#sasl_state.check_password_digest),
server_step(State#sasl_state{mech_mod = Module,
mech_state = MechState},
ClientIn);
_ ->
{error, "no-mechanism"}
end;
false ->
{error, "no-mechanism"}
end.

server_start从sasl_mechanism表中查询出机制模块,并调用机制模块的mech_new/4。这个函数会创建一个机制本身的状态结构。如,PLAIN机制模块的mech_new/4:

1
2
mech_new(_Host, _GetPassword, CheckPassword, _CheckPasswordDigest) ->
{ok, #state{check_password = CheckPassword}}.

server_start函数将这个机制状态保存到SASL状态结构里的mech_state字段,调用server_step。而server_step则调用机制模块的mech_step。

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
server_step(State, ClientIn) ->
Module = State#sasl_state.mech_mod,
MechState = State#sasl_state.mech_state,
case Module:mech_step(MechState, ClientIn) of
{ok, Props} ->
case check_credentials(State, Props) of
ok ->
{ok, Props};
{error, Error} ->
{error, Error}
end;
{ok, Props, ServerOut} ->
case check_credentials(State, Props) of
ok ->
{ok, Props, ServerOut};
{error, Error} ->
{error, Error}
end;
{continue, ServerOut, NewMechState} ->
{continue, ServerOut,
State#sasl_state{mech_state = NewMechState}};
{error, Error, Username} ->
{error, Error, Username};
{error, Error} ->
{error, Error}
end.

Module:mech_step根据自身机制状态,返回不同的值。当验证通过时,返回ok信息。如若还需要其他信息继续验证,则返回continue信息。验证出错时,返回error信息。wait_for_feature_request函数根据不同的返回值,进行不同的处理。

  • 当返回ok信息时,ejabberd_c2s进程向客户端发送验证成功的消息,登录验证流程结束。
  • 当返回continue信息时,表示验证流程需要继续,因而向客户端返回服务端的chelange信息,ejabberd_c2s进程状态变为wait_for_sasl_response。当客户端再回应验证信息后,wait_for_sasl_response函数再次调用server_step进行验证处理。
  • 当返回error信息时,ejabberd_c2s进程向客户端发送验证失败的消息,登录验证流程结束。

PLAIN机制只需要用户名和密码,这些信息附在客户端选择机制的AUTH请求中。因而只需要调用一次mech_step函数,mech_step也因而不会返回continue。cyrsasl_plain:mech_step调用mech_new中传入的check_password回调函数(不同机制会调用不同的回调函数)来检查用户名和密码是否正确。这部分逻辑由ejabberd_auth模块封装不同的模块完成,本文略过。

1
2
3
4
5
6
7
8
9
10
11
12
13
mech_step(State, ClientIn) ->
case prepare(ClientIn) of
[AuthzId, User, Password] ->
case (State#state.check_password)(User, Password) of
{true, AuthModule} ->
{ok, [{username, User}, {authzid, AuthzId},
{auth_module, AuthModule}]};
_ ->
{error, "not-authorized", User}
end;
_ ->
{error, "bad-protocol"}
end.

SCRAM机制需要多次Chelange/Response交互,需要多次调用它的mech_step。因而它在机制状态内部来分步骤完成,具体可以参考cyrsasl_scram.erl,本文不详述。

当机制mech_step返回ok或error时,ejabberd_c2s进程返回给客户端相应的回应,登录验证的流程结束。

注: ejabberd代码版本为2.1.13。