From 43d743e75b7b747341b9a5c36a933b490548bebb Mon Sep 17 00:00:00 2001 From: Karen Arutyunov Date: Sat, 4 Nov 2017 01:17:16 +0300 Subject: Add implementation --- mysql/plugins/pvio/pvio_npipe.c | 383 ++++++++++++++ mysql/plugins/pvio/pvio_shmem.c | 469 +++++++++++++++++ mysql/plugins/pvio/pvio_socket.c | 1070 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 1922 insertions(+) create mode 100644 mysql/plugins/pvio/pvio_npipe.c create mode 100644 mysql/plugins/pvio/pvio_shmem.c create mode 100644 mysql/plugins/pvio/pvio_socket.c (limited to 'mysql/plugins/pvio') diff --git a/mysql/plugins/pvio/pvio_npipe.c b/mysql/plugins/pvio/pvio_npipe.c new file mode 100644 index 0000000..f16beec --- /dev/null +++ b/mysql/plugins/pvio/pvio_npipe.c @@ -0,0 +1,383 @@ +/************************************************************************************ + Copyright (C) 2015 Georg Richter and MariaDB Corporation AB + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Library General Public + License as published by the Free Software Foundation; either + version 2 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Library General Public License for more details. + + You should have received a copy of the GNU Library General Public + License along with this library; if not see + or write to the Free Software Foundation, Inc., + 51 Franklin St., Fifth Floor, Boston, MA 02110, USA + +*************************************************************************************/ + +/* MariaDB virtual IO plugin for Windows named pipe communication */ + +#ifdef _WIN32 + +#include +#include +#include +#include +#include +#include +#include + +/* Function prototypes */ +my_bool pvio_npipe_set_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout); +int pvio_npipe_get_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type); +ssize_t pvio_npipe_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length); +ssize_t pvio_npipe_async_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length); +ssize_t pvio_npipe_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length); +ssize_t pvio_npipe_async_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length); +int pvio_npipe_wait_io_or_timeout(MARIADB_PVIO *pvio, my_bool is_read, int timeout); +my_bool pvio_npipe_blocking(MARIADB_PVIO *pvio, my_bool value, my_bool *old_value); +my_bool pvio_npipe_connect(MARIADB_PVIO *pvio, MA_PVIO_CINFO *cinfo); +my_bool pvio_npipe_close(MARIADB_PVIO *pvio); +int pvio_npipe_fast_send(MARIADB_PVIO *pvio); +int pvio_npipe_keepalive(MARIADB_PVIO *pvio); +my_bool pvio_npipe_get_handle(MARIADB_PVIO *pvio, void *handle); +my_bool pvio_npipe_is_blocking(MARIADB_PVIO *pvio); +int pvio_npipe_shutdown(MARIADB_PVIO *pvio); +my_bool pvio_npipe_is_alive(MARIADB_PVIO *pvio); + +struct st_ma_pvio_methods pvio_npipe_methods= { + pvio_npipe_set_timeout, + pvio_npipe_get_timeout, + pvio_npipe_read, + NULL, + pvio_npipe_write, + NULL, + pvio_npipe_wait_io_or_timeout, + pvio_npipe_blocking, + pvio_npipe_connect, + pvio_npipe_close, + pvio_npipe_fast_send, + pvio_npipe_keepalive, + pvio_npipe_get_handle, + pvio_npipe_is_blocking, + pvio_npipe_is_alive, + NULL, + pvio_npipe_shutdown +}; + +#ifndef HAVE_NPIPE_DYNAMIC +MARIADB_PVIO_PLUGIN pvio_npipe_plugin = +#else +MARIADB_PVIO_PLUGIN _mysql_client_plugin_declaration_ = +#endif +{ + MARIADB_CLIENT_PVIO_PLUGIN, + MARIADB_CLIENT_PVIO_PLUGIN_INTERFACE_VERSION, + "pvio_npipe", + "Georg Richter", + "MariaDB virtual IO plugin for named pipe connection", + {1, 0, 0}, + "LGPL", + NULL, + NULL, + NULL, + NULL, + &pvio_npipe_methods +}; + +struct st_pvio_npipe { + HANDLE pipe; + OVERLAPPED overlapped; + size_t rw_size; + MYSQL *mysql; +}; + +my_bool pvio_npipe_set_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout) +{ + if (!pvio) + return 1; + pvio->timeout[type]= (timeout > 0) ? timeout * 1000 : -1; + return 0; +} + +int pvio_npipe_get_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type) +{ + if (!pvio) + return -1; + return pvio->timeout[type] / 1000; +} + +ssize_t pvio_npipe_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length) +{ + DWORD dwRead= 0; + ssize_t r= -1; + struct st_pvio_npipe *cpipe= NULL; + + if (!pvio || !pvio->data) + return -1; + + cpipe= (struct st_pvio_npipe *)pvio->data; + + if (ReadFile(cpipe->pipe, (LPVOID)buffer, (DWORD)length, &dwRead, &cpipe->overlapped)) + { + r= (ssize_t)dwRead; + goto end; + } + if (GetLastError() == ERROR_IO_PENDING) + { + if (!pvio_npipe_wait_io_or_timeout(pvio, 1, 0)) + r= cpipe->rw_size; + } +end: + return r; +} + +ssize_t pvio_npipe_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length) +{ + DWORD dwWrite= 0; + ssize_t r= -1; + struct st_pvio_npipe *cpipe= NULL; + + if (!pvio || !pvio->data) + return -1; + + cpipe= (struct st_pvio_npipe *)pvio->data; + + if (WriteFile(cpipe->pipe, buffer, (DWORD)length, &dwWrite, &cpipe->overlapped)) + { + r= (ssize_t)dwWrite; + goto end; + } + if (GetLastError() == ERROR_IO_PENDING) + { + if (!pvio_npipe_wait_io_or_timeout(pvio, 0, 0)) + r= cpipe->rw_size; + } +end: + return r; +} + +int pvio_npipe_wait_io_or_timeout(MARIADB_PVIO *pvio, my_bool is_read, int timeout) +{ + int r= -1; + DWORD status; + int save_error; + struct st_pvio_npipe *cpipe= NULL; + + cpipe= (struct st_pvio_npipe *)pvio->data; + + if (!timeout) + timeout= (is_read) ? pvio->timeout[PVIO_READ_TIMEOUT] : pvio->timeout[PVIO_WRITE_TIMEOUT]; + if (!timeout) + timeout= INFINITE; + + status= WaitForSingleObject(cpipe->overlapped.hEvent, timeout); + if (status == WAIT_OBJECT_0) + { + if (GetOverlappedResult(cpipe->pipe, &cpipe->overlapped, (LPDWORD)&cpipe->rw_size, FALSE)) + return 0; + } + /* For other status codes (WAIT_ABANDONED, WAIT_TIMEOUT and WAIT_FAILED) + we return error */ + save_error= GetLastError(); + CancelIo(cpipe->pipe); + SetLastError(save_error); + return -1; +} + +my_bool pvio_npipe_blocking(MARIADB_PVIO *pvio, my_bool block, my_bool *previous_mode) +{ + /* not supported */ + DWORD flags= 0; + struct st_pvio_npipe *cpipe= NULL; + + cpipe= (struct st_pvio_npipe *)pvio->data; + + if (previous_mode) + { + if (!GetNamedPipeHandleState(cpipe->pipe, &flags, NULL, NULL, NULL, NULL, 0)) + return 1; + *previous_mode= flags & PIPE_NOWAIT ? 0 : 1; + } + + flags= block ? PIPE_WAIT : PIPE_NOWAIT; + if (!SetNamedPipeHandleState(cpipe->pipe, &flags, NULL, NULL)) + return 1; + return 0; +} + +int pvio_npipe_keepalive(MARIADB_PVIO *pvio) +{ + /* keep alive is used for TCP/IP connections only */ + return 0; +} + +int pvio_npipe_fast_send(MARIADB_PVIO *pvio) +{ + /* not supported */ + return 0; +} +my_bool pvio_npipe_connect(MARIADB_PVIO *pvio, MA_PVIO_CINFO *cinfo) +{ + struct st_pvio_npipe *cpipe= NULL; + + if (!pvio || !cinfo) + return 1; + + /* if connect timeout is set, we will overwrite read/write timeout */ + if (pvio->timeout[PVIO_CONNECT_TIMEOUT]) + { + pvio->timeout[PVIO_READ_TIMEOUT]= pvio->timeout[PVIO_WRITE_TIMEOUT]= pvio->timeout[PVIO_CONNECT_TIMEOUT]; + } + + if (!(cpipe= (struct st_pvio_npipe *)LocalAlloc(LMEM_ZEROINIT, sizeof(struct st_pvio_npipe)))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_OUT_OF_MEMORY, unknown_sqlstate, 0, ""); + return 1; + } + memset(cpipe, 0, sizeof(struct st_pvio_npipe)); + pvio->data= (void *)cpipe; + cpipe->pipe= INVALID_HANDLE_VALUE; + pvio->mysql= cinfo->mysql; + pvio->type= cinfo->type; + + if (cinfo->type == PVIO_TYPE_NAMEDPIPE) + { + my_bool has_timedout= 0; + char szPipeName[MAX_PATH]; + DWORD dwMode; + + if ( ! cinfo->unix_socket || (cinfo->unix_socket)[0] == 0x00) + cinfo->unix_socket = MARIADB_NAMEDPIPE; + if (!cinfo->host || !strcmp(cinfo->host,LOCAL_HOST)) + cinfo->host=LOCAL_HOST_NAMEDPIPE; + + szPipeName[MAX_PATH - 1]= 0; + snprintf(szPipeName, MAX_PATH - 1, "\\\\%s\\pipe\\%s", cinfo->host, cinfo->unix_socket); + + while (1) + { + if ((cpipe->pipe = CreateFile(szPipeName, + GENERIC_READ | + GENERIC_WRITE, + 0, /* no sharing */ + NULL, /* default security attributes */ + OPEN_EXISTING, + FILE_FLAG_OVERLAPPED, + NULL)) != INVALID_HANDLE_VALUE) + break; + + if (GetLastError() != ERROR_PIPE_BUSY) + { + pvio->set_error(pvio->mysql, CR_NAMEDPIPEOPEN_ERROR, SQLSTATE_UNKNOWN, 0, + cinfo->host, cinfo->unix_socket, GetLastError()); + goto end; + } + + if (has_timedout || !WaitNamedPipe(szPipeName, pvio->timeout[PVIO_CONNECT_TIMEOUT])) + { + pvio->set_error(pvio->mysql, CR_NAMEDPIPEWAIT_ERROR, SQLSTATE_UNKNOWN, 0, + cinfo->host, cinfo->unix_socket, GetLastError()); + goto end; + } + has_timedout= 1; + } + + dwMode = PIPE_READMODE_BYTE | PIPE_WAIT; + if (!SetNamedPipeHandleState(cpipe->pipe, &dwMode, NULL, NULL)) + { + pvio->set_error(pvio->mysql, CR_NAMEDPIPESETSTATE_ERROR, SQLSTATE_UNKNOWN, 0, + cinfo->host, cinfo->unix_socket, (ulong) GetLastError()); + goto end; + } + + /* Register event handler for overlapped IO */ + if (!(cpipe->overlapped.hEvent= CreateEvent(NULL, FALSE, FALSE, NULL))) + { + pvio->set_error(pvio->mysql, CR_EVENT_CREATE_FAILED, SQLSTATE_UNKNOWN, 0, + GetLastError()); + goto end; + } + return 0; + } +end: + if (cpipe) + { + if (cpipe->pipe != INVALID_HANDLE_VALUE) + CloseHandle(cpipe->pipe); + LocalFree(cpipe); + pvio->data= NULL; + } + return 1; +} + +my_bool pvio_npipe_close(MARIADB_PVIO *pvio) +{ + struct st_pvio_npipe *cpipe= NULL; + int r= 0; + + if (!pvio) + return 1; + + if (pvio->data) + { + cpipe= (struct st_pvio_npipe *)pvio->data; + CloseHandle(cpipe->overlapped.hEvent); + if (cpipe->pipe != INVALID_HANDLE_VALUE) + { + CloseHandle(cpipe->pipe); + cpipe->pipe= INVALID_HANDLE_VALUE; + } + LocalFree(pvio->data); + pvio->data= NULL; + } + return r; +} + +my_bool pvio_npipe_get_handle(MARIADB_PVIO *pvio, void *handle) +{ + if (pvio && pvio->data) + { + *(HANDLE *)handle= ((struct st_pvio_npipe *)pvio->data)->pipe; + return 0; + } + return 1; +} + +my_bool pvio_npipe_is_blocking(MARIADB_PVIO *pvio) +{ + DWORD flags= 0; + struct st_pvio_npipe *cpipe= NULL; + + cpipe= (struct st_pvio_npipe *)pvio->data; + + if (!GetNamedPipeHandleState(cpipe->pipe, &flags, NULL, NULL, NULL, NULL, 0)) + return 1; + return (flags & PIPE_NOWAIT) ? 0 : 1; +} + +int pvio_npipe_shutdown(MARIADB_PVIO *pvio) +{ + HANDLE h; + if (pvio_npipe_get_handle(pvio, &h) == 0) + { + return(CancelIoEx(h, NULL) ? 0 : 1); + } + return 1; +} + +my_bool pvio_npipe_is_alive(MARIADB_PVIO *pvio) +{ + HANDLE handle; + if (!pvio || !pvio->data) + return FALSE; + handle= ((struct st_pvio_npipe *)pvio->data)->pipe; + /* Copy data fron named pipe without removing it */ + if (PeekNamedPipe(handle, NULL, 0, NULL, NULL, NULL)) + return TRUE; + return test(GetLastError() != ERROR_BROKEN_PIPE); +} +#endif diff --git a/mysql/plugins/pvio/pvio_shmem.c b/mysql/plugins/pvio/pvio_shmem.c new file mode 100644 index 0000000..09586c6 --- /dev/null +++ b/mysql/plugins/pvio/pvio_shmem.c @@ -0,0 +1,469 @@ +/************************************************************************************ + Copyright (C) 2015 Georg Richter and MariaDB Corporation AB + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Library General Public + License as published by the Free Software Foundation; either + version 2 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Library General Public License for more details. + + You should have received a copy of the GNU Library General Public + License along with this library; if not see + or write to the Free Software Foundation, Inc., + 51 Franklin St., Fifth Floor, Boston, MA 02110, USA + +*************************************************************************************/ +/* MariaDB virtual IO plugin for Windows shared memory communication */ + +#ifdef _WIN32 + +#include +#include +#include +#include +#include +#include +#include + +#define PVIO_SHM_BUFFER_SIZE 16000 + 4 + +my_bool pvio_shm_set_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout); +int pvio_shm_get_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type); +ssize_t pvio_shm_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length); +ssize_t pvio_shm_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length); +int pvio_shm_wait_io_or_timeout(MARIADB_PVIO *pvio, my_bool is_read, int timeout); +my_bool pvio_shm_blocking(MARIADB_PVIO *pvio, my_bool value, my_bool *old_value); +my_bool pvio_shm_connect(MARIADB_PVIO *pvio, MA_PVIO_CINFO *cinfo); +my_bool pvio_shm_close(MARIADB_PVIO *pvio); +int pvio_shm_shutdown(MARIADB_PVIO *pvio); +my_bool pvio_shm_is_alive(MARIADB_PVIO *pvio); +my_bool pvio_shm_get_handle(MARIADB_PVIO *pvio, void *handle); + +struct st_ma_pvio_methods pvio_shm_methods= { + pvio_shm_set_timeout, + pvio_shm_get_timeout, + pvio_shm_read, + NULL, + pvio_shm_write, + NULL, + pvio_shm_wait_io_or_timeout, + pvio_shm_blocking, + pvio_shm_connect, + pvio_shm_close, + NULL, + NULL, + pvio_shm_get_handle, + NULL, + pvio_shm_is_alive, + NULL, + pvio_shm_shutdown +}; + +#ifndef HAVE_SHMEM_DYNAMIC +MARIADB_PVIO_PLUGIN pvio_shmem_plugin= +#else +MARIADB_PVIO_PLUGIN _mysql_client_plugin_declaration_= +#endif +{ + MARIADB_CLIENT_PVIO_PLUGIN, + MARIADB_CLIENT_PVIO_PLUGIN_INTERFACE_VERSION, + "pvio_shmem", + "Georg Richter", + "MariaDB virtual IO plugin for Windows shared memory communication", + {1, 0, 0}, + "LGPPL", + NULL, + NULL, + NULL, + NULL, + &pvio_shm_methods, + +}; + +enum enum_shm_events +{ + PVIO_SHM_SERVER_WROTE= 0, + PVIO_SHM_SERVER_READ, + PVIO_SHM_CLIENT_WROTE, + PVIO_SHM_CLIENT_READ, + PVIO_SHM_CONNECTION_CLOSED +}; + +typedef struct { + HANDLE event[5]; + HANDLE file_map; + LPVOID *map; + char *read_pos; + size_t buffer_size; +} PVIO_SHM; + +char *StrEvent[]= {"SERVER_WROTE", "SERVER_READ", "CLIENT_WROTE", "CLIENT_READ", "CONNECTION_CLOSED"}; + +struct st_pvio_shm { + char *shm_name; +}; + +my_bool pvio_shm_set_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout) +{ + if (!pvio) + return 1; + pvio->timeout[type]= (timeout > 0) ? timeout * 1000 : INFINITE; + return 0; +} + +int pvio_shm_get_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type) +{ + if (!pvio) + return -1; + return pvio->timeout[type] / 1000; +} + +ssize_t pvio_shm_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length) +{ + PVIO_SHM *pvio_shm= (PVIO_SHM *)pvio->data; + size_t copy_size= length; + HANDLE events[2]; + + if (!pvio_shm) + return -1; + + /* we need to wait for write and close events */ + if (!pvio_shm->buffer_size) + { + events[0]= pvio_shm->event[PVIO_SHM_CONNECTION_CLOSED]; + events[1]= pvio_shm->event[PVIO_SHM_SERVER_WROTE]; + + switch(WaitForMultipleObjects(2, events, 0, pvio->timeout[PVIO_READ_TIMEOUT])) + { + case WAIT_OBJECT_0: /* server closed connection */ + SetLastError(ERROR_GRACEFUL_DISCONNECT); + return -1; + case WAIT_OBJECT_0 +1: /* server_wrote event */ + break; + case WAIT_TIMEOUT: + SetLastError(ETIMEDOUT); + default: + return -1; + } + /* server sent data */ + pvio_shm->read_pos= (char *)pvio_shm->map; + pvio_shm->buffer_size= uint4korr(pvio_shm->read_pos); + pvio_shm->read_pos+= 4; + } + + if (pvio_shm->buffer_size < copy_size) + copy_size= pvio_shm->buffer_size; + + if (copy_size) + { + memcpy(buffer, (uchar *)pvio_shm->read_pos, pvio_shm->buffer_size); + pvio_shm->read_pos+= copy_size; + pvio_shm->buffer_size-= copy_size; + } + + /* we need to read again */ + if (!pvio_shm->buffer_size) + if (!SetEvent(pvio_shm->event[PVIO_SHM_CLIENT_READ])) + return -1; + + return (ssize_t)copy_size; +} + +ssize_t pvio_shm_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length) +{ + HANDLE events[2]; + PVIO_SHM *pvio_shm= (PVIO_SHM *)pvio->data; + size_t bytes_to_write= length; + uchar *buffer_pos= (uchar *)buffer; + + if (!pvio_shm) + return -1; + + events[0]= pvio_shm->event[PVIO_SHM_CONNECTION_CLOSED]; + events[1]= pvio_shm->event[PVIO_SHM_SERVER_READ]; + + while (bytes_to_write) + { + size_t pkt_length; + switch (WaitForMultipleObjects(2, events, 0, pvio->timeout[PVIO_WRITE_TIMEOUT])) { + case WAIT_OBJECT_0: /* connection closed */ + SetLastError(ERROR_GRACEFUL_DISCONNECT); + return -1; + case WAIT_OBJECT_0 + 1: /* server_read */ + break; + case WAIT_TIMEOUT: + SetLastError(ETIMEDOUT); + default: + return -1; + } + pkt_length= MIN(PVIO_SHM_BUFFER_SIZE, length); + int4store(pvio_shm->map, pkt_length); + memcpy((uchar *)pvio_shm->map + 4, buffer_pos, length); + buffer_pos+= length; + bytes_to_write-= length; + + if (!SetEvent(pvio_shm->event[PVIO_SHM_CLIENT_WROTE])) + return -1; + } + return (ssize_t)length; +} + + +int pvio_shm_wait_io_or_timeout(MARIADB_PVIO *pvio, my_bool is_read, int timeout) +{ + return 0; +} + +my_bool pvio_shm_blocking(MARIADB_PVIO *pvio, my_bool block, my_bool *previous_mode) +{ + /* not supported */ + return 0; +} + +int pvio_shm_keepalive(MARIADB_PVIO *pvio) +{ + /* not supported */ + return 0; +} + +int pvio_shm_fast_send(MARIADB_PVIO *pvio) +{ + /* not supported */ + return 0; +} + +my_bool pvio_shm_connect(MARIADB_PVIO *pvio, MA_PVIO_CINFO *cinfo) +{ + const char *base_memory_name; + char *prefixes[]= {"", "Global\\", NULL}; + char *shm_name, *shm_suffix, *shm_prefix; + uchar i= 0; + int len; + DWORD cid; + DWORD dwDesiredAccess= EVENT_MODIFY_STATE | SYNCHRONIZE; + HANDLE hdlConnectRequest= NULL, + hdlConnectRequestAnswer= NULL, + file_map= NULL; + LPVOID map= NULL; + PVIO_SHM *pvio_shm= (PVIO_SHM*)LocalAlloc(LMEM_ZEROINIT, sizeof(PVIO_SHM)); + + if (!pvio_shm) + { + PVIO_SET_ERROR(cinfo->mysql, CR_OUT_OF_MEMORY, unknown_sqlstate, 0, ""); + return 0; + } + + /* MariaDB server constructs the event name as follows: + "Global\\base_memory_name" or + "\\base_memory_name" + */ + + + base_memory_name= (cinfo->host) ? cinfo->host : SHM_DEFAULT_NAME; + + if (!(shm_name= (char *)LocalAlloc(LMEM_ZEROINIT, strlen(base_memory_name) + 40))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_OUT_OF_MEMORY, unknown_sqlstate, 0, ""); + goto error; + } + + /* iterate through prefixes */ + while (prefixes[i]) + { + len= sprintf(shm_name, "%s%s_", prefixes[i], base_memory_name); + shm_suffix= shm_name + len; + strcpy(shm_suffix, "CONNECT_REQUEST"); + if ((hdlConnectRequest= OpenEvent(dwDesiredAccess, 0, shm_name))) + { + /* save prefix to prevent further loop */ + shm_prefix= prefixes[i]; + break; + } + i++; + } + if (!hdlConnectRequest) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Opening CONNECT_REQUEST event failed", GetLastError()); + goto error; + } + + strcpy(shm_suffix, "CONNECT_ANSWER"); + if (!(hdlConnectRequestAnswer= OpenEvent(dwDesiredAccess, 0, shm_name))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Opening CONNECT_ANSWER event failed", GetLastError()); + goto error; + } + + /* get connection id, so we can build the filename used for connection */ + strcpy(shm_suffix, "CONNECT_DATA"); + if (!(file_map= OpenFileMapping(FILE_MAP_WRITE, 0, shm_name))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "OpenFileMapping failed", GetLastError()); + goto error; + } + + /* try to get first 4 bytes, which represents connection_id */ + if (!(map= MapViewOfFile(file_map, FILE_MAP_WRITE, 0, 0, sizeof(cid)))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Reading connection_id failed", GetLastError()); + goto error; + } + + /* notify server */ + if (!SetEvent(hdlConnectRequest)) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Failed sending connection request", GetLastError()); + goto error; + } + + /* Wait for server answer */ + switch(WaitForSingleObject(hdlConnectRequestAnswer, pvio->timeout[PVIO_CONNECT_TIMEOUT])) { + case WAIT_ABANDONED: + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Mutex was not released in time", GetLastError()); + goto error; + break; + case WAIT_FAILED: + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Operation wait failed", GetLastError()); + goto error; + break; + case WAIT_TIMEOUT: + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Operation timed out", GetLastError()); + goto error; + break; + case WAIT_OBJECT_0: + break; + default: + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Wait for server failed", GetLastError()); + break; + } + + cid= uint4korr(map); + + len= sprintf(shm_name, "%s%s_%d_", shm_prefix, base_memory_name, cid); + shm_suffix= shm_name + len; + + strcpy(shm_suffix, "DATA"); + pvio_shm->file_map= OpenFileMapping(FILE_MAP_WRITE, 0, shm_name); + if (pvio_shm->file_map == NULL) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "OpenFileMapping failed", GetLastError()); + goto error; + } + if (!(pvio_shm->map= MapViewOfFile(pvio_shm->file_map, FILE_MAP_WRITE, 0, 0, PVIO_SHM_BUFFER_SIZE))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "MapViewOfFile failed", GetLastError()); + goto error; + } + + for (i=0; i < 5; i++) + { + strcpy(shm_suffix, StrEvent[i]); + if (!(pvio_shm->event[i]= OpenEvent(dwDesiredAccess, 0, shm_name))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SHARED_MEMORY_CONNECT_ERROR, unknown_sqlstate, 0, "Couldn't create event", GetLastError()); + goto error; + } + } + /* we will first read from server */ + SetEvent(pvio_shm->event[PVIO_SHM_SERVER_READ]); + +error: + if (hdlConnectRequest) + CloseHandle(hdlConnectRequest); + if (hdlConnectRequestAnswer) + CloseHandle(hdlConnectRequestAnswer); + if (shm_name) + LocalFree(shm_name); + if (map) + UnmapViewOfFile(map); + if (file_map) + CloseHandle(file_map); + if (pvio_shm) + { + /* check if all events are set */ + if (pvio_shm->event[4]) + { + pvio->data= (void *)pvio_shm; + pvio->mysql= cinfo->mysql; + pvio->type= cinfo->type; + pvio_shm->read_pos= (char *)pvio_shm->map; + pvio->mysql->net.pvio= pvio; + return 0; + } + for (i=0;i < 5; i++) + if (pvio_shm->event[i]) + CloseHandle(pvio_shm->event[i]); + if (pvio_shm->map) + UnmapViewOfFile(pvio_shm->map); + if (pvio_shm->file_map) + CloseHandle(pvio_shm->file_map); + LocalFree(pvio_shm); + } + return 1; + +} + +my_bool pvio_shm_close(MARIADB_PVIO *pvio) +{ + PVIO_SHM *pvio_shm= (PVIO_SHM *)pvio->data; + int i; + + if (!pvio_shm) + return 1; + + /* notify server */ + SetEvent(pvio_shm->event[PVIO_SHM_CONNECTION_CLOSED]); + + UnmapViewOfFile(pvio_shm->map); + CloseHandle(pvio_shm->file_map); + + for (i=0; i < 5; i++) + CloseHandle(pvio_shm->event[i]); + + LocalFree(pvio_shm); + pvio->data= NULL; + return 0; +} + +my_bool pvio_shm_get_socket(MARIADB_PVIO *pvio, void *handle) +{ + return 1; +} + +my_bool pvio_shm_is_blocking(MARIADB_PVIO *pvio) +{ + return 1; +} + +int pvio_shm_shutdown(MARIADB_PVIO *pvio) +{ + PVIO_SHM *pvio_shm= (PVIO_SHM *)pvio->data; + if (pvio_shm) + return (SetEvent(pvio_shm->event[PVIO_SHM_CONNECTION_CLOSED]) ? 0 : 1); + return 1; +} + +my_bool pvio_shm_is_alive(MARIADB_PVIO *pvio) +{ + PVIO_SHM *pvio_shm; + if (!pvio || !pvio->data) + return FALSE; + pvio_shm= (PVIO_SHM *)pvio->data; + return WaitForSingleObject(pvio_shm->event[PVIO_SHM_CONNECTION_CLOSED], 0)!=WAIT_OBJECT_0; +} + +my_bool pvio_shm_get_handle(MARIADB_PVIO *pvio, void *handle) +{ + + *(HANDLE **)handle= 0; + if (!pvio || !pvio->data) + return FALSE; + *(HANDLE **)handle= ((PVIO_SHM*)pvio->data)->event; + return TRUE; +} +#endif + diff --git a/mysql/plugins/pvio/pvio_socket.c b/mysql/plugins/pvio/pvio_socket.c new file mode 100644 index 0000000..737ca15 --- /dev/null +++ b/mysql/plugins/pvio/pvio_socket.c @@ -0,0 +1,1070 @@ +/************************************************************************************ + Copyright (C) 2015,2016 MariaDB Corporation AB, + + This library is free software; you can redistribute it and/or + modify it under the terms of the GNU Library General Public + License as published by the Free Software Foundation; either + version 2 of the License, or (at your option) any later version. + + This library is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Library General Public License for more details. + + You should have received a copy of the GNU Library General Public + License along with this library; if not see + or write to the Free Software Foundation, Inc., + 51 Franklin St., Fifth Floor, Boston, MA 02110, USA +*************************************************************************************/ + +/* + MariaDB virtual IO plugin for socket communication: + + The plugin handles connections via unix and network sockets. it is enabled by + default and compiled into Connector/C. +*/ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#ifndef _WIN32 +#ifdef HAVE_SYS_UN_H +#include +#endif +#ifdef HAVE_POLL +#include +#endif +#ifdef HAVE_SYS_IOCTL_H +#include +#endif +#ifdef HAVE_FCNTL_H +#include +#endif +#include +#include +#include +#include +#include +#define IS_SOCKET_EINTR(err) (err == SOCKET_EINTR) +#else +#include +#define O_NONBLOCK 1 +#define MSG_DONTWAIT 0 +#define IS_SOCKET_EINTR(err) 0 +#endif + +#ifndef SOCKET_ERROR +#define SOCKET_ERROR -1 +#endif + +#define DNS_TIMEOUT 30 + +#ifndef O_NONBLOCK +#if defined(O_NDELAY) +#define O_NONBLOCK O_NODELAY +#elif defined (O_FNDELAY) +#define O_NONBLOCK O_FNDELAY +#else +#error socket blocking is not supported on this platform +#endif +#endif + + +/* Function prototypes */ +my_bool pvio_socket_set_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout); +int pvio_socket_get_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type); +ssize_t pvio_socket_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length); +ssize_t pvio_socket_async_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length); +ssize_t pvio_socket_async_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length); +ssize_t pvio_socket_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length); +int pvio_socket_wait_io_or_timeout(MARIADB_PVIO *pvio, my_bool is_read, int timeout); +my_bool pvio_socket_blocking(MARIADB_PVIO *pvio, my_bool value, my_bool *old_value); +my_bool pvio_socket_connect(MARIADB_PVIO *pvio, MA_PVIO_CINFO *cinfo); +my_bool pvio_socket_close(MARIADB_PVIO *pvio); +int pvio_socket_fast_send(MARIADB_PVIO *pvio); +int pvio_socket_keepalive(MARIADB_PVIO *pvio); +my_bool pvio_socket_get_handle(MARIADB_PVIO *pvio, void *handle); +my_bool pvio_socket_is_blocking(MARIADB_PVIO *pvio); +my_bool pvio_socket_is_alive(MARIADB_PVIO *pvio); +my_bool pvio_socket_has_data(MARIADB_PVIO *pvio, ssize_t *data_len); +int pvio_socket_shutdown(MARIADB_PVIO *pvio); + +static int pvio_socket_init(char *unused1, + size_t unused2, + int unused3, + va_list); +static int pvio_socket_end(void); +static ssize_t ma_send(my_socket socket, const uchar *buffer, size_t length, int flags); +static ssize_t ma_recv(my_socket socket, uchar *buffer, size_t length, int flags); + +struct st_ma_pvio_methods pvio_socket_methods= { + pvio_socket_set_timeout, + pvio_socket_get_timeout, + pvio_socket_read, + pvio_socket_async_read, + pvio_socket_write, + pvio_socket_async_write, + pvio_socket_wait_io_or_timeout, + pvio_socket_blocking, + pvio_socket_connect, + pvio_socket_close, + pvio_socket_fast_send, + pvio_socket_keepalive, + pvio_socket_get_handle, + pvio_socket_is_blocking, + pvio_socket_is_alive, + pvio_socket_has_data, + pvio_socket_shutdown +}; + +#ifndef HAVE_SOCKET_DYNAMIC +MARIADB_PVIO_PLUGIN pvio_socket_plugin= +#else +MARIADB_PVIO_PLUGIN _mysql_client_plugin_declaration_ +#endif +{ + MARIADB_CLIENT_PVIO_PLUGIN, + MARIADB_CLIENT_PVIO_PLUGIN_INTERFACE_VERSION, + "pvio_socket", + "Georg Richter", + "MariaDB virtual IO plugin for socket communication", + {1, 0, 0}, + "LGPL", + NULL, + &pvio_socket_init, + &pvio_socket_end, + NULL, + &pvio_socket_methods +}; + +struct st_pvio_socket { + my_socket socket; + int fcntl_mode; + MYSQL *mysql; +}; + +static my_bool pvio_socket_initialized= FALSE; + +static int pvio_socket_init(char *errmsg __attribute__((unused)), + size_t errmsg_length __attribute__((unused)), + int unused __attribute__((unused)), + va_list va __attribute__((unused))) +{ + pvio_socket_initialized= TRUE; + return 0; +} + +static int pvio_socket_end(void) +{ + if (!pvio_socket_initialized) + return 1; + return 0; +} + +my_bool pvio_socket_change_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout) +{ + struct timeval tm; + struct st_pvio_socket *csock= NULL; + if (!pvio) + return 1; + if (!(csock= (struct st_pvio_socket *)pvio->data)) + return 1; + tm.tv_sec= timeout / 1000; + tm.tv_usec= (timeout % 1000) * 1000; + switch(type) + { + case PVIO_WRITE_TIMEOUT: +#ifndef _WIN32 + setsockopt(csock->socket, SOL_SOCKET, SO_SNDTIMEO, (const char *)&tm, sizeof(tm)); +#else + setsockopt(csock->socket, SOL_SOCKET, SO_SNDTIMEO, (const char *)&timeout, sizeof(int)); +#endif + break; + case PVIO_READ_TIMEOUT: +#ifndef _WIN32 + setsockopt(csock->socket, SOL_SOCKET, SO_RCVTIMEO, (const char *)&tm, sizeof(tm)); +#else + setsockopt(csock->socket, SOL_SOCKET, SO_RCVTIMEO, (const char *)&timeout, sizeof(int)); +#endif + break; + default: + break; + } + return 0; +} + +/* {{{ pvio_socket_set_timeout */ +/* + set timeout value + + SYNOPSIS + pvio_socket_set_timeout + pvio PVIO + type timeout type (connect, read, write) + timeout timeout in seconds + + DESCRIPTION + Sets timeout values for connection-, read or write time out. + PVIO internally stores all timeout values in milliseconds, but + accepts and returns all time values in seconds (like api does). + + RETURNS + 0 Success + 1 Error +*/ +my_bool pvio_socket_set_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type, int timeout) +{ + struct st_pvio_socket *csock= NULL; + if (!pvio) + return 1; + csock= (struct st_pvio_socket *)pvio->data; + pvio->timeout[type]= (timeout > 0) ? timeout * 1000 : -1; + if (csock) + return pvio_socket_change_timeout(pvio, type, timeout * 1000); + return 0; +} +/* }}} */ + +/* {{{ pvio_socket_get_timeout */ +/* + get timeout value + + SYNOPSIS + pvio_socket_get_timeout + pvio PVIO + type timeout type (connect, read, write) + + DESCRIPTION + Returns timeout values for connection-, read or write time out. + PVIO internally stores all timeout values in milliseconds, but + accepts and returns all time values in seconds (like api does). + + RETURNS + 0...n time out value + -1 error +*/ +int pvio_socket_get_timeout(MARIADB_PVIO *pvio, enum enum_pvio_timeout type) +{ + if (!pvio) + return -1; + return pvio->timeout[type] / 1000; +} +/* }}} */ + +/* {{{ pvio_socket_read */ +/* + read from socket + + SYNOPSIS + pvio_socket_read() + pvio PVIO + buffer read buffer + length buffer length + + DESCRIPTION + reads up to length bytes into specified buffer. In the event of an + error erno is set to indicate it. + + RETURNS + 1..n number of bytes read + 0 peer has performed shutdown + -1 on error + +*/ +ssize_t pvio_socket_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length) +{ + ssize_t r; + int read_flags= MSG_DONTWAIT; + struct st_pvio_socket *csock; + int timeout; + + if (!pvio || !pvio->data) + return -1; + + csock= (struct st_pvio_socket *)pvio->data; + timeout = pvio->timeout[PVIO_READ_TIMEOUT]; + + while ((r = ma_recv(csock->socket, (void *)buffer, length, read_flags)) == -1) + { + int err = socket_errno; + if ((err != SOCKET_EAGAIN && err != SOCKET_EWOULDBLOCK) || timeout == 0) + return r; + + if (pvio_socket_wait_io_or_timeout(pvio, TRUE, timeout) < 1) + return -1; + } + return r; +} +/* }}} */ + +/* {{{ pvio_socket_async_read */ +/* + read from socket + + SYNOPSIS + pvio_socket_async_read() + pvio PVIO + buffer read buffer + length buffer length + + DESCRIPTION + reads up to length bytes into specified buffer. In the event of an + error erno is set to indicate it. + + RETURNS + 1..n number of bytes read + 0 peer has performed shutdown + -1 on error + +*/ +ssize_t pvio_socket_async_read(MARIADB_PVIO *pvio, uchar *buffer, size_t length) +{ + ssize_t r= -1; +#ifndef _WIN32 + int read_flags= MSG_DONTWAIT; +#endif + struct st_pvio_socket *csock= NULL; + + if (!pvio || !pvio->data) + return -1; + + csock= (struct st_pvio_socket *)pvio->data; + +#ifndef _WIN32 + r= recv(csock->socket,(void *)buffer, length, read_flags); +#else + /* Windows doesn't support MSG_DONTWAIT, so we need to set + socket to non blocking */ + pvio_socket_blocking(pvio, 0, 0); + r= recv(csock->socket, (char *)buffer, (int)length, 0); +#endif + return r; +} +/* }}} */ + +static ssize_t ma_send(my_socket socket, const uchar *buffer, size_t length, int flags) +{ + ssize_t r; +#if !defined(MSG_NOSIGNAL) && !defined(SO_NOSIGPIPE) && !defined(_WIN32) + struct sigaction act, oldact; + act.sa_handler= SIG_IGN; + sigaction(SIGPIPE, &act, &oldact); +#endif + do { + r = send(socket, buffer, IF_WIN((int)length,length), flags); + } + while (r == -1 && IS_SOCKET_EINTR(socket_errno)); +#if !defined(MSG_NOSIGNAL) && !defined(SO_NOSIGPIPE) && !defined(_WIN32) + sigaction(SIGPIPE, &oldact, NULL); +#endif + return r; +} + +static ssize_t ma_recv(my_socket socket, uchar *buffer, size_t length, int flags) +{ + ssize_t r; + do { + r = recv(socket, buffer, IF_WIN((int)length, length), flags); + } + while (r == -1 && IS_SOCKET_EINTR(socket_errno)); + return r; +} + +/* {{{ pvio_socket_async_write */ +/* + write to socket + + SYNOPSIS + pvio_socket_async_write() + pvio PVIO + buffer read buffer + length buffer length + + DESCRIPTION + writes up to length bytes to socket. In the event of an + error erno is set to indicate it. + + RETURNS + 1..n number of bytes read + 0 peer has performed shutdown + -1 on error + +*/ +ssize_t pvio_socket_async_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length) +{ + ssize_t r= -1; + struct st_pvio_socket *csock= NULL; +#ifndef _WIN32 + int write_flags= MSG_DONTWAIT; +#ifdef MSG_NOSIGNAL + write_flags|= MSG_NOSIGNAL; +#endif +#endif + + if (!pvio || !pvio->data) + return -1; + + csock= (struct st_pvio_socket *)pvio->data; + +#ifndef WIN32 + r= ma_send(csock->socket, buffer, length, write_flags); +#else + /* Windows doesn't support MSG_DONTWAIT, so we need to set + socket to non blocking */ + pvio_socket_blocking(pvio, 0, 0); + r= send(csock->socket, buffer, (int)length, 0); +#endif + + return r; +} +/* }}} */ + +/* {{{ pvio_socket_write */ +/* + write to socket + + SYNOPSIS + pvio_socket_write() + pvio PVIO + buffer read buffer + length buffer length + + DESCRIPTION + writes up to length bytes to socket. In the event of an + error erno is set to indicate it. + + RETURNS + 1..n number of bytes read + 0 peer has performed shutdown + -1 on error + +*/ +ssize_t pvio_socket_write(MARIADB_PVIO *pvio, const uchar *buffer, size_t length) +{ + ssize_t r; + struct st_pvio_socket *csock; + int timeout; + int send_flags= MSG_DONTWAIT; +#ifdef MSG_NOSIGNAL + send_flags|= MSG_NOSIGNAL; +#endif + if (!pvio || !pvio->data) + return -1; + + csock= (struct st_pvio_socket *)pvio->data; + timeout = pvio->timeout[PVIO_WRITE_TIMEOUT]; + + while ((r = ma_send(csock->socket, (void *)buffer, length,send_flags)) == -1) + { + int err = socket_errno; + if ((err != SOCKET_EAGAIN && err != SOCKET_EWOULDBLOCK)|| timeout == 0) + return r; + if (pvio_socket_wait_io_or_timeout(pvio, FALSE, timeout) < 1) + return -1; + } + return r; +} +/* }}} */ + +int pvio_socket_wait_io_or_timeout(MARIADB_PVIO *pvio, my_bool is_read, int timeout) +{ + int rc; + struct st_pvio_socket *csock= NULL; + +#ifndef _WIN32 + struct pollfd p_fd; +#else + struct timeval tv= {0,0}; + fd_set fds, exc_fds; +#endif + + if (!pvio || !pvio->data) + return 0; + + csock= (struct st_pvio_socket *)pvio->data; + { +#ifndef _WIN32 + memset(&p_fd, 0, sizeof(p_fd)); + p_fd.fd= csock->socket; + p_fd.events= (is_read) ? POLLIN : POLLOUT; + + do { + rc= poll(&p_fd, 1, timeout); + } while (rc == -1 && errno == EINTR); + + if (rc == 0) + errno= ETIMEDOUT; +#else + FD_ZERO(&fds); + FD_ZERO(&exc_fds); + + FD_SET(csock->socket, &fds); + FD_SET(csock->socket, &exc_fds); + + if (timeout >= 0) + { + tv.tv_sec= timeout / 1000; + tv.tv_usec= (timeout % 1000) * 1000; + } + + rc= select(0, (is_read) ? &fds : NULL, + (is_read) ? NULL : &fds, + &exc_fds, + (timeout >= 0) ? &tv : NULL); + + if (rc == SOCKET_ERROR) + { + errno= WSAGetLastError(); + } + else if (rc == 0) + { + rc= SOCKET_ERROR; + WSASetLastError(WSAETIMEDOUT); + errno= ETIMEDOUT; + } + else if (FD_ISSET(csock->socket, &exc_fds)) + { + int err; + int len = sizeof(int); + if (getsockopt(csock->socket, SOL_SOCKET, SO_ERROR, (char *)&err, &len) != SOCKET_ERROR) + { + WSASetLastError(err); + errno= err; + } + rc= SOCKET_ERROR; + } + +#endif + } + return rc; +} + +my_bool pvio_socket_blocking(MARIADB_PVIO *pvio, my_bool block, my_bool *previous_mode) +{ + my_bool is_blocking; + struct st_pvio_socket *csock; + int new_fcntl_mode; + + if (!pvio || !pvio->data) + return 1; + + csock = (struct st_pvio_socket *)pvio->data; + + is_blocking = !(csock->fcntl_mode & O_NONBLOCK); + if (previous_mode) + *previous_mode = is_blocking; + + if (is_blocking == block) + return 0; + + if (block) + new_fcntl_mode = csock->fcntl_mode & ~O_NONBLOCK; + else + new_fcntl_mode = csock->fcntl_mode | O_NONBLOCK; + +#ifdef _WIN32 + { + ulong arg = block ? 0 : 1; + if (ioctlsocket(csock->socket, FIONBIO, (void *)&arg)) + { + return(WSAGetLastError()); + } + } +#else + if (fcntl(csock->socket, F_SETFL, new_fcntl_mode) == -1) + { + return errno; + } +#endif + csock->fcntl_mode = new_fcntl_mode; + return 0; +} + +static int pvio_socket_internal_connect(MARIADB_PVIO *pvio, + const struct sockaddr *name, + size_t namelen) +{ + int rc= 0; + struct st_pvio_socket *csock= NULL; + int timeout; + + if (!pvio || !pvio->data) + return 1; + + csock= (struct st_pvio_socket *)pvio->data; + timeout= pvio->timeout[PVIO_CONNECT_TIMEOUT]; + + /* set non blocking */ + pvio_socket_blocking(pvio, 0, 0); + +#ifndef _WIN32 + do { + rc= connect(csock->socket, (struct sockaddr*) name, (int)namelen); + } while (rc == -1 && errno == EINTR); + /* in case a timeout values was set we need to check error values + EINPROGRESS and EAGAIN */ + if (timeout != 0 && rc == -1 && + (errno == EINPROGRESS || errno == EAGAIN)) + { + rc= pvio_socket_wait_io_or_timeout(pvio, FALSE, timeout); + if (rc < 1) + return -1; + { + int error; + socklen_t error_len= sizeof(error); + if ((rc = getsockopt(csock->socket, SOL_SOCKET, SO_ERROR, + (char *)&error, &error_len)) < 0) + return errno; + else if (error) + return error; + } + } +#ifdef __APPLE__ + if (csock->socket) + { + int val= 1; + setsockopt(csock->socket, SOL_SOCKET, SO_NOSIGPIPE, (void *)&val, sizeof(int)); + } +#endif +#else + rc= connect(csock->socket, (struct sockaddr*) name, (int)namelen); + if (rc == SOCKET_ERROR) + { + if (WSAGetLastError() == WSAEWOULDBLOCK) + { + if (pvio_socket_wait_io_or_timeout(pvio, FALSE, timeout) < 0) + return -1; + rc= 0; + } + } +#endif + return rc; +} + +int pvio_socket_keepalive(MARIADB_PVIO *pvio) +{ + int opt= 1; + struct st_pvio_socket *csock= NULL; + + if (!pvio || !pvio->data) + return 1; + + csock= (struct st_pvio_socket *)pvio->data; + + return setsockopt(csock->socket, SOL_SOCKET, SO_KEEPALIVE, +#ifndef _WIN32 + (const void *)&opt, sizeof(opt)); +#else + (char *)&opt, (int)sizeof(opt)); +#endif +} + +int pvio_socket_fast_send(MARIADB_PVIO *pvio) +{ + int r= 0; + struct st_pvio_socket *csock= NULL; + + if (!pvio || !pvio->data) + return 1; + + csock= (struct st_pvio_socket *)pvio->data; + +/* Setting IP_TOS is not recommended on Windows. See + http://msdn.microsoft.com/en-us/library/windows/desktop/ms738586(v=vs.85).aspx +*/ +#if !defined(_WIN32) && defined(IPTOS_THROUGHPUT) + { + int tos = IPTOS_THROUGHPUT; + r= setsockopt(csock->socket, IPPROTO_IP, IP_TOS, + (const void *)&tos, sizeof(tos)); + } +#endif /* !_WIN32 && IPTOS_THROUGHPUT */ + if (!r) + { + int opt = 1; + /* turn off nagle algorithm */ + r= setsockopt(csock->socket, IPPROTO_TCP, TCP_NODELAY, +#ifdef _WIN32 + (const char *)&opt, (int)sizeof(opt)); +#else + (const void *)&opt, sizeof(opt)); +#endif + } + return r; +} + +static int +pvio_socket_connect_sync_or_async(MARIADB_PVIO *pvio, + const struct sockaddr *name, uint namelen) +{ + MYSQL *mysql= pvio->mysql; + if (mysql->options.extension && mysql->options.extension->async_context && + mysql->options.extension->async_context->active) + { + /* even if we are not connected yet, application needs to check socket + * via mysql_get_socket api call, so we need to assign pvio */ + mysql->options.extension->async_context->pvio= pvio; + pvio_socket_blocking(pvio, 0, 0); + return my_connect_async(pvio, name, namelen, pvio->timeout[PVIO_CONNECT_TIMEOUT]); + } + + return pvio_socket_internal_connect(pvio, name, namelen); +} + +my_bool pvio_socket_connect(MARIADB_PVIO *pvio, MA_PVIO_CINFO *cinfo) +{ + struct st_pvio_socket *csock= NULL; + MYSQL *mysql; + + if (!pvio || !cinfo) + return 1; + + if (!(csock= (struct st_pvio_socket *)calloc(1, sizeof(struct st_pvio_socket)))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_OUT_OF_MEMORY, unknown_sqlstate, 0, ""); + return 1; + } + pvio->data= (void *)csock; + csock->socket= -1; + mysql= pvio->mysql= cinfo->mysql; + pvio->type= cinfo->type; + + if (cinfo->type == PVIO_TYPE_UNIXSOCKET) + { +#ifndef _WIN32 +#ifdef HAVE_SYS_UN_H + struct sockaddr_un UNIXaddr; + if ((csock->socket = socket(AF_UNIX,SOCK_STREAM,0)) == SOCKET_ERROR) + { + PVIO_SET_ERROR(cinfo->mysql, CR_SOCKET_CREATE_ERROR, unknown_sqlstate, 0, errno); + goto error; + } + memset((char*) &UNIXaddr, 0, sizeof(UNIXaddr)); + UNIXaddr.sun_family = AF_UNIX; + strcpy(UNIXaddr.sun_path, cinfo->unix_socket); + if (pvio_socket_connect_sync_or_async(pvio, (struct sockaddr *) &UNIXaddr, + sizeof(UNIXaddr))) + { + PVIO_SET_ERROR(cinfo->mysql, CR_CONNECTION_ERROR, SQLSTATE_UNKNOWN, + ER(CR_CONNECTION_ERROR), cinfo->unix_socket, socket_errno); + goto error; + } + if (pvio_socket_blocking(pvio, 1, 0) == SOCKET_ERROR) + { + goto error; + } +#else +/* todo: error, not supported */ +#endif +#endif + } else if (cinfo->type == PVIO_TYPE_SOCKET) + { + struct addrinfo hints, *save_res= 0, *bind_res= 0, *res= 0, *bres= 0; + char server_port[NI_MAXSERV]; + int gai_rc; + int rc= 0; + time_t start_t= time(NULL); +#ifdef _WIN32 + DWORD wait_gai; +#else + unsigned int wait_gai; +#endif + + memset(&server_port, 0, NI_MAXSERV); + snprintf(server_port, NI_MAXSERV, "%d", cinfo->port); + + /* set hints for getaddrinfo */ + memset(&hints, 0, sizeof(hints)); + hints.ai_protocol= IPPROTO_TCP; /* TCP connections only */ + hints.ai_family= AF_UNSPEC; /* includes: IPv4, IPv6 or hostname */ + hints.ai_socktype= SOCK_STREAM; + + /* if client has multiple interfaces, we will bind socket to given + * bind_address */ + if (cinfo->mysql->options.bind_address) + { + wait_gai= 1; + while ((gai_rc= getaddrinfo(cinfo->mysql->options.bind_address, 0, + &hints, &bind_res)) == EAI_AGAIN) + { + unsigned int timeout= mysql->options.connect_timeout ? + mysql->options.connect_timeout : DNS_TIMEOUT; + if (time(NULL) - start_t > (time_t)timeout) + break; +#ifndef _WIN32 + usleep(wait_gai); +#else + Sleep(wait_gai); +#endif + wait_gai*= 2; + } + if (gai_rc != 0 || !bind_res) + { + PVIO_SET_ERROR(cinfo->mysql, CR_BIND_ADDR_FAILED, SQLSTATE_UNKNOWN, + CER(CR_BIND_ADDR_FAILED), cinfo->mysql->options.bind_address, gai_rc); + goto error; + } + } + /* Get the address information for the server using getaddrinfo() */ + wait_gai= 1; + while ((gai_rc= getaddrinfo(cinfo->host, server_port, + &hints, &res)) == EAI_AGAIN) + { + unsigned int timeout= mysql->options.connect_timeout ? + mysql->options.connect_timeout : DNS_TIMEOUT; + if (time(NULL) - start_t > (time_t)timeout) + break; +#ifndef _WIN32 + usleep(wait_gai); +#else + Sleep(wait_gai); +#endif + wait_gai*= 2; + } + if (gai_rc != 0 || !res) + { + PVIO_SET_ERROR(cinfo->mysql, CR_UNKNOWN_HOST, SQLSTATE_UNKNOWN, + ER(CR_UNKNOWN_HOST), cinfo->host, gai_rc); + if (bind_res) + freeaddrinfo(bind_res); + goto error; + } + + /* res is a linked list of addresses for the given hostname. We loop until + we are able to connect to one address or all connect attempts failed */ + for (save_res= res; save_res; save_res= save_res->ai_next) + { + csock->socket= socket(save_res->ai_family, save_res->ai_socktype, + save_res->ai_protocol); + if (csock->socket == SOCKET_ERROR) + /* Errors will be handled after loop finished */ + continue; + + if (bind_res) + { + for (bres= bind_res; bres; bres= bres->ai_next) + { + if (!(rc= bind(csock->socket, bres->ai_addr, (int)bres->ai_addrlen))) + break; + } + if (rc) + { + closesocket(csock->socket); + continue; + } + } + + rc= pvio_socket_connect_sync_or_async(pvio, save_res->ai_addr, (uint)save_res->ai_addrlen); + if (!rc) + { + MYSQL *mysql= pvio->mysql; + if (mysql->options.extension && mysql->options.extension->async_context && + mysql->options.extension->async_context->active) + break; + if (pvio_socket_blocking(pvio, 0, 0) == SOCKET_ERROR) + { + closesocket(csock->socket); + continue; + } + break; /* success! */ + } + } + + freeaddrinfo(res); + if (bind_res) + freeaddrinfo(bind_res); + + if (csock->socket == SOCKET_ERROR) + { + PVIO_SET_ERROR(cinfo->mysql, CR_IPSOCK_ERROR, SQLSTATE_UNKNOWN, ER(CR_IPSOCK_ERROR), + socket_errno); + goto error; + } + + /* last call to connect 2 failed */ + if (rc) + { + PVIO_SET_ERROR(cinfo->mysql, CR_CONNECTION_ERROR, SQLSTATE_UNKNOWN, + ER(CR_CONN_HOST_ERROR), cinfo->host, +#ifdef _WIN32 + errno); +#else + socket_errno); +#endif + goto error; + } + if (pvio_socket_blocking(pvio, 1, 0) == SOCKET_ERROR) + goto error; + } + /* apply timeouts */ + if (pvio->timeout[PVIO_CONNECT_TIMEOUT] > 0) + { + pvio_socket_change_timeout(pvio, PVIO_READ_TIMEOUT, pvio->timeout[PVIO_CONNECT_TIMEOUT]); + pvio_socket_change_timeout(pvio, PVIO_WRITE_TIMEOUT, pvio->timeout[PVIO_CONNECT_TIMEOUT]); + } + else + { + if (pvio->timeout[PVIO_WRITE_TIMEOUT] > 0) + pvio_socket_change_timeout(pvio, PVIO_WRITE_TIMEOUT, pvio->timeout[PVIO_WRITE_TIMEOUT]); + if (pvio->timeout[PVIO_READ_TIMEOUT] > 0) + pvio_socket_change_timeout(pvio, PVIO_READ_TIMEOUT, pvio->timeout[PVIO_READ_TIMEOUT]); + } + return 0; +error: + /* close socket: MDEV-10891 */ + if (csock->socket != -1) + closesocket(csock->socket); + if (pvio->data) + { + free((gptr)pvio->data); + pvio->data= NULL; + } + return 1; +} + +/* {{{ my_bool pvio_socket_close() */ +my_bool pvio_socket_close(MARIADB_PVIO *pvio) +{ + struct st_pvio_socket *csock= NULL; + int r= 0; + + if (!pvio) + return 1; + + if (pvio->data) + { + csock= (struct st_pvio_socket *)pvio->data; + if (csock && csock->socket != -1) + { + r= shutdown(csock->socket ,2); + r= closesocket(csock->socket); + csock->socket= -1; + } + free((gptr)pvio->data); + pvio->data= NULL; + } + return r; +} +/* }}} */ + +/* {{{ my_socket pvio_socket_get_handle */ +my_bool pvio_socket_get_handle(MARIADB_PVIO *pvio, void *handle) +{ + if (pvio && pvio->data && handle) + { + *(my_socket *)handle= ((struct st_pvio_socket *)pvio->data)->socket; + return 0; + } + return 1; +} +/* }}} */ + +/* {{{ my_bool pvio_socket_is_blocking(MARIADB_PVIO *pvio) */ +my_bool pvio_socket_is_blocking(MARIADB_PVIO *pvio) +{ + struct st_pvio_socket *csock= NULL; + my_bool r; + + if (!pvio || !pvio->data) + return 0; + + csock= (struct st_pvio_socket *)pvio->data; + r = !(csock->fcntl_mode & O_NONBLOCK); + return r; +} +/* }}} */ + +/* {{{ my_bool pvio_socket_is_alive(MARIADB_PVIO *pvio) */ +my_bool pvio_socket_is_alive(MARIADB_PVIO *pvio) +{ + struct st_pvio_socket *csock= NULL; + #ifndef _WIN32 + struct pollfd poll_fd; +#else + FD_SET sfds; + struct timeval tv= {0,0}; +#endif + int res; + + if (!pvio || !pvio->data) + return 0; + + csock= (struct st_pvio_socket *)pvio->data; +#ifndef _WIN32 + memset(&poll_fd, 0, sizeof(struct pollfd)); + poll_fd.events= POLLPRI | POLLIN; + poll_fd.revents= POLLERR; + poll_fd.fd= csock->socket; + + res= poll(&poll_fd, 1, 0); + if (res <= 0) /* timeout or error */ + return FALSE; + if (!(poll_fd.revents & POLLERR)) + return FALSE; + if (!(poll_fd.revents & (POLLIN | POLLPRI))) + return FALSE; + return TRUE; +#else + /* We can't use the WSAPoll function, it's broken :-( + (see Windows 8 Bugs 309411 - WSAPoll does not report failed connections) + Instead we need to use select function: + If TIMEVAL is initialized to {0, 0}, select will return immediately; + this is used to poll the state of the selected sockets. + */ + FD_ZERO(&sfds); + FD_SET(csock->socket, &sfds); + + res= select((int)csock->socket + 1, &sfds, NULL, NULL, &tv); + if (res > 0 && FD_ISSET(csock->socket, &sfds)) + return TRUE; + return FALSE; +#endif +} +/* }}} */ + +/* {{{ my_boool pvio_socket_has_data */ +my_bool pvio_socket_has_data(MARIADB_PVIO *pvio, ssize_t *data_len) +{ + struct st_pvio_socket *csock= NULL; + char tmp_buf; + ssize_t len; + my_bool mode; + + if (!pvio || !pvio->data) + return 0; + + csock= (struct st_pvio_socket *)pvio->data; + /* MSG_PEEK: Peeks at the incoming data. The data is copied into the buffer, + but is not removed from the input queue. + */ + pvio_socket_blocking(pvio, 0, &mode); + len= recv(csock->socket, &tmp_buf, sizeof(tmp_buf), MSG_PEEK); + pvio_socket_blocking(pvio, mode, 0); + if (len < 0) + return 1; + *data_len= len; + return 0; +} +/* }}} */ + +int pvio_socket_shutdown(MARIADB_PVIO *pvio) +{ + if (pvio && pvio->data) + { + my_socket s = ((struct st_pvio_socket *)pvio->data)->socket; +#ifdef _WIN32 + shutdown(s, SD_BOTH); + CancelIoEx((HANDLE)s, NULL); +#else + shutdown(s, SHUT_RDWR); +#endif + } + return -1; +} -- cgit v1.1