Line data Source code
1 : /* TDSPool - Connection pooling for TDS based databases
2 : * Copyright (C) 2001, 2002, 2003, 2004, 2005 Brian Bruns
3 : *
4 : * This program is free software; you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation; either version 2 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program; if not, write to the Free Software
16 : * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17 : *
18 : */
19 :
20 : #include <config.h>
21 :
22 : #include <stdarg.h>
23 : #include <stdio.h>
24 : #include <errno.h>
25 : #include <assert.h>
26 :
27 : #if HAVE_STDLIB_H
28 : #include <stdlib.h>
29 : #endif /* HAVE_STDLIB_H */
30 :
31 : #if HAVE_STRING_H
32 : #include <string.h>
33 : #endif /* HAVE_STRING_H */
34 :
35 : #if HAVE_UNISTD_H
36 : #include <unistd.h>
37 : #endif /* HAVE_UNISTD_H */
38 :
39 : #if HAVE_SYS_TYPES_H
40 : #include <sys/types.h>
41 : #endif /* HAVE_SYS_TYPES_H */
42 :
43 : #if HAVE_SYS_SOCKET_H
44 : #include <sys/socket.h>
45 : #endif /* HAVE_SYS_SOCKET_H */
46 :
47 : #include "pool.h"
48 : #include <freetds/server.h>
49 : #include <freetds/utils/string.h>
50 :
51 : static TDS_POOL_USER *pool_user_find_new(TDS_POOL * pool);
52 : static bool pool_user_login(TDS_POOL * pool, TDS_POOL_USER * puser);
53 : static bool pool_user_read(TDS_POOL * pool, TDS_POOL_USER * puser);
54 : static void login_execute(TDS_POOL_EVENT *base_event);
55 : static void end_login_execute(TDS_POOL_EVENT *base_event);
56 :
57 : void
58 2 : pool_user_init(TDS_POOL * pool)
59 : {
60 4 : dlist_user_init(&pool->users);
61 4 : dlist_user_init(&pool->waiters);
62 2 : pool->ctx = tds_alloc_context(NULL);
63 2 : }
64 :
65 : void
66 2 : pool_user_destroy(TDS_POOL * pool)
67 : {
68 4 : while (dlist_user_first(&pool->users))
69 0 : pool_free_user(pool, dlist_user_first(&pool->users));
70 2 : while (dlist_user_first(&pool->waiters))
71 0 : pool_free_user(pool, dlist_user_first(&pool->waiters));
72 :
73 2 : tds_free_context(pool->ctx);
74 2 : pool->ctx = NULL;
75 2 : }
76 :
77 : static TDS_POOL_USER *
78 782 : pool_user_find_new(TDS_POOL * pool)
79 : {
80 : TDS_POOL_USER *puser;
81 :
82 : /* did we exhaust the number of concurrent users? */
83 782 : if (pool->num_users >= MAX_POOL_USERS) {
84 0 : fprintf(stderr, "Max concurrent users exceeded, increase in pool.h\n");
85 0 : return NULL;
86 : }
87 :
88 782 : puser = tds_new0(TDS_POOL_USER, 1);
89 782 : if (!puser) {
90 0 : fprintf(stderr, "Out of memory\n");
91 0 : return NULL;
92 : }
93 :
94 782 : dlist_user_append(&pool->users, puser);
95 782 : pool->num_users++;
96 :
97 782 : return puser;
98 : }
99 :
100 : typedef struct {
101 : TDS_POOL_EVENT common;
102 : TDS_POOL *pool;
103 : TDS_POOL_USER *puser;
104 : bool success;
105 : } LOGIN_EVENT;
106 :
107 782 : static TDS_THREAD_PROC_DECLARE(login_proc, arg)
108 : {
109 782 : LOGIN_EVENT *ev = (LOGIN_EVENT *) arg;
110 :
111 782 : ev->success = pool_user_login(ev->pool, ev->puser);
112 :
113 782 : pool_event_add(ev->pool, &ev->common, login_execute);
114 782 : return TDS_THREAD_RESULT(0);
115 : }
116 :
117 : static void
118 782 : login_execute(TDS_POOL_EVENT *base_event)
119 : {
120 782 : LOGIN_EVENT *ev = (LOGIN_EVENT *) base_event;
121 782 : TDS_POOL_USER *puser = ev->puser;
122 782 : TDS_POOL *pool = ev->pool;
123 :
124 782 : if (!ev->success) {
125 : /* login failed...free socket */
126 2 : pool_free_user(pool, puser);
127 2 : return;
128 : }
129 :
130 780 : puser->sock.poll_recv = true;
131 :
132 : /* try to assign a member, connection can have transactions
133 : * and so on so deassign only when disconnected */
134 780 : pool_user_query(pool, puser);
135 :
136 780 : tdsdump_log(TDS_DBG_INFO1, "user state %d\n", puser->user_state);
137 :
138 780 : assert(puser->login || puser->user_state == TDS_SRV_QUERY);
139 : }
140 :
141 :
142 : /**
143 : * Check if an error returned by accept() is permanent.
144 : * \param err socket error code
145 : * \return true if the listening socket cannot be used anymore
146 : */
147 : static inline bool
148 : accept_error_is_fatal(int err)
149 : {
150 : #ifdef _WIN32
151 : return err == WSAEINVAL || err == WSAENOTSOCK || err == WSAEBADF;
152 : #else
153 0 : return err == EINVAL || err == ENOTSOCK || err == EBADF;
154 : #endif
155 : }
156 :
157 : /*
158 : * pool_user_create
159 : * accepts a client connection and adds it to the users list and returns it
160 : */
161 : TDS_POOL_USER *
162 782 : pool_user_create(TDS_POOL * pool, TDS_SYS_SOCKET s)
163 : {
164 : TDS_POOL_USER *puser;
165 : TDS_SYS_SOCKET fd;
166 : TDSSOCKET *tds;
167 : LOGIN_EVENT *ev;
168 :
169 782 : tdsdump_log(TDS_DBG_NETWORK, "accepting connection\n");
170 782 : if (TDS_IS_SOCKET_INVALID(fd = tds_accept(s, NULL, NULL))) {
171 0 : int err = sock_errno;
172 0 : char *errstr = sock_strerror(err);
173 :
174 0 : tdsdump_log(TDS_DBG_ERROR, "error calling accept: %s\n", errstr);
175 : /*
176 : * These errors mean the listening socket is not usable
177 : * (for instance not a listening socket), retrying would
178 : * just loop forever.
179 : */
180 0 : if (accept_error_is_fatal(err)) {
181 0 : fprintf(stderr, "Error: accept failed, %s\n", errstr);
182 0 : exit(EXIT_FAILURE);
183 : }
184 : sock_strerror_free(errstr);
185 : return NULL;
186 : }
187 :
188 782 : tds_socket_set_nodelay(fd);
189 782 : tds_socket_set_nosigpipe(fd, 1);
190 :
191 782 : if (tds_socket_set_nonblocking(fd) != 0) {
192 0 : CLOSESOCKET(fd);
193 0 : return NULL;
194 : }
195 :
196 782 : puser = pool_user_find_new(pool);
197 782 : if (!puser) {
198 0 : CLOSESOCKET(fd);
199 0 : return NULL;
200 : }
201 :
202 782 : tds = tds_alloc_socket(pool->ctx, BLOCKSIZ);
203 782 : if (!tds) {
204 0 : CLOSESOCKET(fd);
205 0 : return NULL;
206 : }
207 782 : ev = tds_new0(LOGIN_EVENT, 1);
208 782 : if (!ev || TDS_FAILED(tds_iconv_open(tds->conn, "UTF-8", 0))) {
209 0 : free(ev);
210 0 : tds_free_socket(tds);
211 0 : CLOSESOCKET(fd);
212 0 : return NULL;
213 : }
214 782 : tds_set_s(tds, fd);
215 782 : tds->state = TDS_IDLE;
216 782 : tds->out_flag = TDS_LOGIN;
217 :
218 782 : puser->sock.tds = tds;
219 782 : puser->user_state = TDS_SRV_QUERY;
220 782 : puser->sock.poll_recv = false;
221 782 : puser->sock.poll_send = false;
222 :
223 : /* launch login asyncronously */
224 782 : ev->puser = puser;
225 782 : ev->pool = pool;
226 :
227 782 : if (tds_thread_create_detached(login_proc, ev) != 0) {
228 0 : pool_free_user(pool, puser);
229 0 : fprintf(stderr, "error creating thread\n");
230 0 : return NULL;
231 : }
232 :
233 : return puser;
234 : }
235 :
236 : /*
237 : * pool_free_user
238 : * close out a disconnected user.
239 : */
240 : void
241 782 : pool_free_user(TDS_POOL *pool, TDS_POOL_USER * puser)
242 : {
243 782 : TDS_POOL_MEMBER *pmbr = puser->assigned_member;
244 782 : if (pmbr) {
245 778 : assert(pmbr->current_user == puser);
246 778 : pool_deassign_member(pool, pmbr);
247 778 : pool_reset_member(pool, pmbr);
248 : }
249 :
250 782 : tds_free_socket(puser->sock.tds);
251 782 : tds_free_login(puser->login);
252 :
253 : /* make sure to decrement the waiters list if he is waiting */
254 782 : if (puser->user_state == TDS_SRV_WAIT)
255 0 : dlist_user_remove(&pool->waiters, puser);
256 : else
257 782 : dlist_user_remove(&pool->users, puser);
258 782 : pool->num_users--;
259 782 : free(puser);
260 782 : }
261 :
262 : /*
263 : * pool_process_users
264 : * check the fd_set for user input, allocate a pool member to it, and forward
265 : * the query to that member.
266 : */
267 : void
268 36185 : pool_process_users(TDS_POOL * pool, struct pollfd *fds, unsigned num_fds)
269 : {
270 : TDS_POOL_USER *puser, *next;
271 : short revents;
272 :
273 142760 : for (next = dlist_user_first(&pool->users); (puser = next) != NULL; ) {
274 :
275 140780 : next = dlist_user_next(&pool->users, puser);
276 :
277 70390 : if (!puser->sock.tds || puser->sock.poll_index >= num_fds)
278 0 : continue; /* dead connection */
279 :
280 70390 : revents = fds[puser->sock.poll_index].revents;
281 :
282 70390 : if (puser->sock.poll_recv && (revents & (POLLIN|POLLHUP)) != 0) {
283 16964 : assert(puser->user_state == TDS_SRV_QUERY);
284 16964 : if (!pool_user_read(pool, puser))
285 778 : continue;
286 : }
287 69612 : if (puser->sock.poll_send && (revents & POLLOUT) != 0) {
288 0 : if (!pool_write_data(&puser->assigned_member->sock, &puser->sock))
289 0 : pool_free_member(pool, puser->assigned_member);
290 : }
291 : } /* for */
292 36185 : }
293 :
294 : /*
295 : * pool_user_login
296 : * Reads clients login packet and forges a login acknowledgement sequence
297 : */
298 : static bool
299 782 : pool_user_login(TDS_POOL * pool, TDS_POOL_USER * puser)
300 : {
301 : TDSSOCKET *tds;
302 : TDSLOGIN *login;
303 :
304 782 : tds = puser->sock.tds;
305 1564 : while (tds->in_len <= tds->in_pos)
306 782 : if (tds_read_packet(tds) < 0)
307 : return false;
308 :
309 782 : tdsdump_log(TDS_DBG_NETWORK, "got packet type %d\n", tds->in_flag);
310 782 : if (tds->in_flag == TDS71_PRELOGIN) {
311 782 : if (!tds->conn->tds_version)
312 782 : tds->conn->tds_version = 0x701;
313 782 : tds->out_flag = TDS_REPLY;
314 : // TODO proper one !!
315 : // TODO detect TDS version here ??
316 782 : tds_put_n(tds, "\x00\x00\x1a\x00\x06" /* version */
317 : "\x01\x00\x20\x00\x01" /* encryption */
318 : "\x02\x00\x21\x00\x01" /* instance ?? */
319 : "\x03\x00\x22\x00\x00" /* process id ?? */
320 : "\x04\x00\x22\x00\x01" /* MARS */
321 : "\xff"
322 : "\x0a\x00\x06\x40\x00\x00"
323 : "\x02"
324 : "\x00"
325 : ""
326 : "\x00", 0x23);
327 782 : tds_flush_packet(tds);
328 :
329 : /* read another packet */
330 782 : tds->in_pos = tds->in_len;
331 1564 : while (tds->in_len <= tds->in_pos)
332 782 : if (tds_read_packet(tds) < 0)
333 : return false;
334 : }
335 :
336 782 : puser->login = login = tds_alloc_login(true);
337 782 : if (!login) {
338 0 : tdsdump_log(TDS_DBG_ERROR, "tds_alloc_login() failed.\n");
339 : return false;
340 : }
341 782 : if (tds->in_flag == TDS_LOGIN) {
342 0 : if (!tds->conn->tds_version)
343 0 : tds->conn->tds_version = 0x500;
344 0 : tds_read_login(tds, login);
345 782 : } else if (tds->in_flag == TDS7_LOGIN) {
346 782 : if (!tds->conn->tds_version)
347 0 : tds->conn->tds_version = 0x700;
348 782 : if (!tds7_read_login(tds, login))
349 : return false;
350 : } else {
351 : return false;
352 : }
353 :
354 : /* check we support version required */
355 : // TODO function to check it
356 782 : if (!IS_TDS71_PLUS(login))
357 : return false;
358 :
359 782 : tds->in_len = tds->in_pos = 0;
360 :
361 782 : dump_login(login);
362 1564 : if (strcmp(tds_dstr_cstr(&login->user_name), pool->user) != 0
363 1564 : || strcmp(tds_dstr_cstr(&login->password), pool->password) != 0)
364 : /* TODO send nack before exiting */
365 : return false;
366 :
367 : return true;
368 : }
369 :
370 : bool
371 780 : pool_user_send_login_ack(TDS_POOL * pool, TDS_POOL_USER * puser)
372 : {
373 : char msg[256];
374 : char block[32];
375 780 : TDSSOCKET *tds = puser->sock.tds, *mtds = puser->assigned_member->sock.tds;
376 780 : TDSLOGIN *login = puser->login;
377 : const char *database;
378 780 : const char *server = mtds->conn->server ? mtds->conn->server : "JDBC";
379 : bool dbname_mismatch, odbc_mismatch;
380 :
381 780 : pool->user_logins++;
382 :
383 : /* copy a bit of information, resize socket with block */
384 780 : tds->conn->tds_version = mtds->conn->tds_version;
385 780 : tds->conn->product_version = mtds->conn->product_version;
386 780 : memcpy(tds->conn->collation, mtds->conn->collation, sizeof(tds->conn->collation));
387 780 : tds->conn->tds71rev1 = mtds->conn->tds71rev1;
388 780 : free(tds->conn->product_name);
389 780 : tds->conn->product_name = strdup(mtds->conn->product_name);
390 780 : tds_realloc_socket(tds, mtds->conn->env.block_size);
391 780 : tds->conn->env.block_size = mtds->conn->env.block_size;
392 780 : tds->conn->client_spid = mtds->conn->spid;
393 :
394 : /* if database is different use USE statement */
395 780 : database = pool->database;
396 1560 : dbname_mismatch = !tds_dstr_isempty(&login->database)
397 1090 : && strcasecmp(tds_dstr_cstr(&login->database), database) != 0;
398 780 : odbc_mismatch = (login->option_flag2 & TDS_ODBC_ON) == 0;
399 780 : if (dbname_mismatch || odbc_mismatch) {
400 : char *str;
401 396 : int len = 128 + tds_quote_id(mtds, NULL, tds_dstr_cstr(&login->database),-1);
402 : TDSRET ret;
403 :
404 198 : if ((str = tds_new(char, len)) == NULL)
405 : return false;
406 :
407 198 : str[0] = 0;
408 : /* swicth to dblib options */
409 198 : if (odbc_mismatch)
410 196 : strcat(str, "SET ANSI_DEFAULTS OFF\nSET CONCAT_NULL_YIELDS_NULL OFF\n");
411 198 : if (dbname_mismatch) {
412 2 : strcat(str, "USE ");
413 4 : tds_quote_id(mtds, strchr(str, 0), tds_dstr_cstr(&login->database), -1);
414 : }
415 198 : ret = tds_submit_query(mtds, str);
416 198 : free(str);
417 198 : if (TDS_FAILED(ret) || TDS_FAILED(tds_process_simple_query(mtds)))
418 : return false;
419 198 : if (dbname_mismatch)
420 4 : database = tds_dstr_cstr(&login->database);
421 : else
422 196 : database = mtds->conn->env.database;
423 : }
424 :
425 : // 7.0
426 : // env database
427 : // database change message (with server name correct)
428 : // env language
429 : // language change message
430 : // env 0x3 charset ("iso_1")
431 : // env 0x5 lcid ("1033")
432 : // env 0x6 ("196609" ?? 0x30001)
433 : // loginack
434 : // env 0x4 packet size
435 : // done
436 : //
437 : // 7.1/7.2/7.3
438 : // env database
439 : // database change message (with server name correct)
440 : // env 0x7 collation
441 : // env language
442 : // language change message
443 : // loginack
444 : // env 0x4 packet size
445 : // done
446 780 : tds->out_flag = TDS_REPLY;
447 780 : tds_env_change(tds, TDS_ENV_DATABASE, "master", database);
448 780 : sprintf(msg, "Changed database context to '%s'.", database);
449 780 : tds_send_msg(tds, 5701, 2, 0, msg, server, NULL, 1);
450 780 : if (!login->suppress_language) {
451 780 : tds_env_change(tds, TDS_ENV_LANG, NULL, "us_english");
452 780 : tds_send_msg(tds, 5703, 1, 0, "Changed language setting to 'us_english'.", server, NULL, 1);
453 : }
454 :
455 780 : if (IS_TDS71_PLUS(tds->conn)) {
456 780 : tds_put_byte(tds, TDS_ENVCHANGE_TOKEN);
457 780 : tds_put_smallint(tds, 8);
458 780 : tds_put_byte(tds, TDS_ENV_SQLCOLLATION);
459 780 : tds_put_byte(tds, 5);
460 780 : tds_put_n(tds, tds->conn->collation, 5);
461 780 : tds_put_byte(tds, 0);
462 : }
463 :
464 780 : tds_send_login_ack(tds, mtds->conn->product_name);
465 780 : sprintf(block, "%d", tds->conn->env.block_size);
466 780 : tds_env_change(tds, TDS_ENV_PACKSIZE, block, block);
467 : /* tds_send_capabilities_token(tds); */
468 780 : tds_send_done_token(tds, TDS_DONE_FINAL, 0);
469 :
470 : /* send it! */
471 780 : tds_flush_packet(tds);
472 :
473 780 : tds_free_login(login);
474 780 : puser->login = NULL;
475 780 : return true;
476 : }
477 :
478 : /*
479 : * pool_user_read
480 : * checks the packet type of data coming from the client and allocates a
481 : * pool member if necessary.
482 : */
483 : static bool
484 16964 : pool_user_read(TDS_POOL * pool, TDS_POOL_USER * puser)
485 : {
486 16964 : TDSSOCKET *tds = puser->sock.tds;
487 16964 : TDS_POOL_MEMBER *pmbr = NULL;
488 :
489 : for (;;) {
490 : TDS_UCHAR in_flag;
491 :
492 33941 : if (pool_packet_read(tds))
493 : break;
494 17755 : if (tds->in_len == 0) {
495 778 : tdsdump_log(TDS_DBG_INFO1, "user disconnected\n");
496 778 : pool_free_user(pool, puser);
497 778 : return false;
498 : }
499 :
500 16977 : tdsdump_dump_buf(TDS_DBG_NETWORK, "Got packet from client:", tds->in_buf, tds->in_len);
501 :
502 16977 : in_flag = tds->in_buf[0];
503 : switch (in_flag) {
504 16977 : case TDS_QUERY:
505 : case TDS_NORMAL:
506 : case TDS_RPC:
507 : case TDS_BULK:
508 : case TDS_CANCEL:
509 : case TDS7_TRANS:
510 16977 : if (!pool_write_data(&puser->sock, &puser->assigned_member->sock)) {
511 0 : pool_reset_member(pool, puser->assigned_member);
512 0 : return false;
513 : }
514 16977 : pmbr = puser->assigned_member;
515 : break;
516 :
517 0 : default:
518 0 : tdsdump_log(TDS_DBG_ERROR, "Unrecognized packet type, closing user\n");
519 0 : pool_free_user(pool, puser);
520 0 : return false;
521 : }
522 16977 : if (tds->in_pos < tds->in_len)
523 : /* partial write, schedule a future write */
524 : break;
525 : }
526 16186 : if (pmbr && !pmbr->sock.poll_send)
527 16184 : tds_connection_flush(pmbr->sock.tds);
528 : return true;
529 : }
530 :
531 : void
532 780 : pool_user_query(TDS_POOL * pool, TDS_POOL_USER * puser)
533 : {
534 : TDS_POOL_MEMBER *pmbr;
535 :
536 780 : tdsdump_log(TDS_DBG_FUNC, "pool_user_query\n");
537 :
538 780 : assert(puser->assigned_member == NULL);
539 780 : assert(puser->login);
540 :
541 780 : puser->user_state = TDS_SRV_QUERY;
542 780 : pmbr = pool_assign_idle_member(pool, puser);
543 780 : if (!pmbr) {
544 : /*
545 : * put into wait state
546 : * check when member is deallocated
547 : */
548 0 : tdsdump_log(TDS_DBG_INFO1, "Not enough free members...placing user in WAIT\n");
549 0 : puser->user_state = TDS_SRV_WAIT;
550 0 : puser->sock.poll_recv = false;
551 0 : puser->sock.poll_send = false;
552 0 : dlist_user_remove(&pool->users, puser);
553 0 : dlist_user_append(&pool->waiters, puser);
554 : }
555 780 : }
556 :
557 : typedef struct {
558 : TDS_POOL_EVENT common;
559 : TDS_POOL *pool;
560 : TDS_POOL_USER *puser;
561 : bool success;
562 : } END_LOGIN_EVENT;
563 :
564 757 : static TDS_THREAD_PROC_DECLARE(end_login_proc, arg)
565 : {
566 757 : END_LOGIN_EVENT *ev = (END_LOGIN_EVENT *) arg;
567 757 : TDS_POOL *pool = ev->pool;
568 :
569 757 : ev->success = pool_user_send_login_ack(pool, ev->puser);
570 :
571 757 : pool_event_add(pool, &ev->common, end_login_execute);
572 757 : return TDS_THREAD_RESULT(0);
573 : }
574 :
575 : static void
576 757 : end_login_execute(TDS_POOL_EVENT *base_event)
577 : {
578 757 : END_LOGIN_EVENT *ev = (END_LOGIN_EVENT *) base_event;
579 757 : TDS_POOL *pool = ev->pool;
580 757 : TDS_POOL_USER *puser = ev->puser;
581 757 : TDS_POOL_MEMBER *pmbr = puser->assigned_member;
582 :
583 757 : if (!ev->success) {
584 0 : pool_free_member(pool, pmbr);
585 0 : return;
586 : }
587 :
588 757 : puser->sock.poll_recv = true;
589 757 : puser->sock.poll_send = false;
590 757 : pmbr->sock.poll_recv = true;
591 757 : pmbr->sock.poll_send = false;
592 : }
593 :
594 : /**
595 : * Handle async login
596 : */
597 : void
598 757 : pool_user_finish_login(TDS_POOL * pool, TDS_POOL_USER * puser)
599 : {
600 757 : END_LOGIN_EVENT *ev = tds_new0(END_LOGIN_EVENT, 1);
601 757 : if (!ev) {
602 0 : pool_free_member(pool, puser->assigned_member);
603 0 : return;
604 : }
605 :
606 757 : ev->pool = pool;
607 757 : ev->puser = puser;
608 :
609 757 : if (tds_thread_create_detached(end_login_proc, ev) != 0) {
610 0 : pool_free_member(pool, puser->assigned_member);
611 0 : free(ev);
612 0 : fprintf(stderr, "error creating thread\n");
613 : }
614 : }
|
