cover/mysql_conn.COVER.html

1 %% MySQL/OTP – MySQL client library for Erlang/OTP
2 %% Copyright (C) 2014-2021 Viktor Söderqvist
3 %%
4 %% This file is part of MySQL/OTP.
5 %%
6 %% MySQL/OTP is free software: you can redistribute it and/or modify it under
7 %% the terms of the GNU Lesser General Public License as published by the Free
8 %% Software Foundation, either version 3 of the License, or (at your option)
9 %% any later version.
10 %%
11 %% This program is distributed in the hope that it will be useful, but WITHOUT
12 %% ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 %% FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 %% more details.
15 %%
16 %% You should have received a copy of the GNU Lesser General Public License
17 %% along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19 %% @doc This module implements parts of the MySQL client/server protocol.
20 %%
21 %% The protocol is described in the document "MySQL Internals" which can be
22 %% found under "MySQL Documentation: Expert Guides" on http://dev.mysql.com/.
23 %%
24 %% TCP communication is not handled in this module. Most of the public functions
25 %% take funs for data communitaction as parameters.
26 %% @private
27 -module(mysql_conn).
28
29 -behaviour(gen_server).
30 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
31 code_change/3, format_status/1]).
32
33 -if(?OTP_RELEASE < 25).
34 -export([format_status/2]).
35 -endif.
36
37 -define(default_host, "localhost").
38 -define(default_port, 3306).
39 -define(default_user, <<>>).
40 -define(default_password, <<>>).
41 -define(default_connect_timeout, 5000).
42 -define(default_query_timeout, infinity).
43 -define(default_query_cache_time, 60000). %% for query/3.
44 -define(default_ping_timeout, 60000).
45
46 -define(cmd_timeout, 3000). %% Timeout used for various commands to the server
47
48 %% Errors that cause "implicit rollback"
49 -define(ERROR_DEADLOCK, 1213).
50
51 %% --- Gen_server callbacks ---
52
53 -include("records.hrl").
54 -include("server_status.hrl").
55
56 %% Gen_server state
57 -record(state, {server_version, connection_id, socket, sockmod, tcp_opts, ssl_opts,
58 host, port, user, password, database, queries, prepares,
59 auth_plugin_name, auth_plugin_data, allowed_local_paths,
60 log_warnings, log_slow_queries,
61 connect_timeout, ping_timeout, query_timeout, query_cache_time,
62 affected_rows = 0, status = 0, warning_count = 0, insert_id = 0,
63 transaction_levels = [], ping_ref = undefined,
64 stmts = #{}, query_cache = mysql_cache:new(), cap_found_rows = false,
65 float_as_decimal = false, decode_decimal = auto}).
66
67 %% @private
68 init(Opts) ->
69 48 ok = logger:set_process_metadata(#{domain => [mysql]}),
70
71 %% Connect
72 48 Host = proplists:get_value(host, Opts, ?default_host),
73
74 48 DefaultPort = case Host of
75 2 {local, _LocalAddr} -> 0;
76 46 _NonLocalAddr -> ?default_port
77 end,
78 48 Port = proplists:get_value(port, Opts, DefaultPort),
79
80 48 User = proplists:get_value(user, Opts, ?default_user),
81 48 Password = proplists:get_value(password, Opts, ?default_password),
82 48 Database = proplists:get_value(database, Opts, undefined),
83 48 AllowedLocalPaths = proplists:get_value(allowed_local_paths, Opts, []),
84 48 LogWarn = proplists:get_value(log_warnings, Opts, true),
85 48 LogSlow = proplists:get_value(log_slow_queries, Opts, false),
86 48 KeepAlive = proplists:get_value(keep_alive, Opts, false),
87 48 ConnectTimeout = proplists:get_value(connect_timeout, Opts,
88 ?default_connect_timeout),
89 48 QueryTimeout = proplists:get_value(query_timeout, Opts,
90 ?default_query_timeout),
91 48 QueryCacheTime = proplists:get_value(query_cache_time, Opts,
92 ?default_query_cache_time),
93 48 TcpOpts = proplists:get_value(tcp_options, Opts, []),
94 48 SetFoundRows = proplists:get_value(found_rows, Opts, false),
95 48 SSLOpts = proplists:get_value(ssl, Opts, undefined),
96
97 48 Queries = proplists:get_value(queries, Opts, []),
98 48 Prepares = proplists:get_value(prepare, Opts, []),
99 48 FloatAsDecimal = proplists:get_value(float_as_decimal, Opts, false),
100 48 DecodeDecimal = proplists:get_value(decode_decimal, Opts, auto),
101
102 48 true = lists:all(fun mysql_protocol:valid_path/1, AllowedLocalPaths),
103
104 48 PingTimeout = case KeepAlive of
105 4 true -> ?default_ping_timeout;
106 43 false -> infinity;
107 1 N when N > 0 -> N
108 end,
109
110 48 State0 = #state{
111 tcp_opts = TcpOpts,
112 ssl_opts = SSLOpts,
113 host = Host, port = Port,
114 user = User, password = Password,
115 database = Database,
116 allowed_local_paths = AllowedLocalPaths,
117 queries = Queries, prepares = Prepares,
118 log_warnings = LogWarn, log_slow_queries = LogSlow,
119 connect_timeout = ConnectTimeout,
120 ping_timeout = PingTimeout,
121 query_timeout = QueryTimeout,
122 query_cache_time = QueryCacheTime,
123 cap_found_rows = (SetFoundRows =:= true),
124 float_as_decimal = FloatAsDecimal,
125 decode_decimal = DecodeDecimal
126 },
127
128 48 case proplists:get_value(connect_mode, Opts, synchronous) of
129 synchronous ->
130 45 case connect(State0) of
131 {ok, State1} ->
132 42 {ok, State1};
133 {error, Reason} ->
134 3 {stop, Reason}
135 end;
136 asynchronous ->
137 2 gen_server:cast(self(), connect),
138 2 {ok, State0};
139 lazy ->
140 1 {ok, State0}
141 end.
142
143 connect(#state{connect_timeout = ConnectTimeout} = State) ->
144 48 MainPid = self(),
145 48 Pid = spawn_link(
146 fun () ->
147 48 {ok, State1}=connect_socket(State),
148 48 case handshake(State1) of
149 {ok, #state{sockmod = SockMod, socket = Socket} = State2} ->
150 46 SockMod:controlling_process(Socket, MainPid),
151 46 MainPid ! {self(), {ok, State2}};
152 {error, _} = E ->
153 2 MainPid ! {self(), E}
154 end
155 end
156 ),
157 48 receive
158 {Pid, {ok, State3}} ->
159 46 post_connect(State3);
160 {Pid, {error, _} = E} ->
161 2 E
162 after ConnectTimeout ->
163 
:-(
 unlink(Pid),
164 
:-(
 exit(Pid, kill),
165 
:-(
 {error, timeout}
166 end.
167
168 connect_socket(#state{tcp_opts = TcpOpts, host = Host, port = Port} = State) ->
169 %% Connect socket
170 48 SockOpts = sanitize_tcp_opts(Host, TcpOpts),
171 48 {ok, Socket} = gen_tcp:connect(Host, Port, SockOpts),
172
173 %% If buffer wasn't specifically defined make it at least as
174 %% large as recbuf, as suggested by the inet:setopts() docs.
175 48 case proplists:is_defined(buffer, TcpOpts) of
176 true ->
177 
:-(
 ok;
178 false ->
179 48 {ok, [{buffer, Buffer}]} = inet:getopts(Socket, [buffer]),
180 48 {ok, [{recbuf, Recbuf}]} = inet:getopts(Socket, [recbuf]),
181 48 ok = inet:setopts(Socket, [{buffer, max(Buffer, Recbuf)}])
182 end,
183
184 48 {ok, State#state{socket = Socket}}.
185
186 sanitize_tcp_opts(Host, [{inet_backend, _} = InetBackend | TcpOpts0]) ->
187 %% This option is be used to turn on the experimental socket backend for
188 %% gen_tcp/inet (OTP/23). If given, it must remain the first option in the
189 %% list.
190 2 [InetBackend | sanitize_tcp_opts(Host, TcpOpts0)];
191 sanitize_tcp_opts(Host, TcpOpts0) ->
192 48 NodelaySupported = not is_tuple(Host) orelse element(1, Host) =/= local,
193 48 TcpOpts1 = lists:filter(
194 fun
195 
:-(
 ({mode, _}) -> false;
196 
:-(
 (binary) -> false;
197 
:-(
 (list) -> false;
198 
:-(
 ({packet, _}) -> false;
199 
:-(
 ({active, _}) -> false;
200 
:-(
 ({nodelay, _}) -> NodelaySupported;
201 
:-(
 (_) -> true
202 end,
203 TcpOpts0
204 ),
205 48 TcpOpts2 = case NodelaySupported andalso not lists:keymember(nodelay, 1, TcpOpts1) of
206 46 true -> [{nodelay, true} | TcpOpts1];
207 2 false -> TcpOpts1
208 end,
209 48 [binary, {packet, raw}, {active, false} | TcpOpts2].
210
211 handshake(#state{socket = Socket0, ssl_opts = SSLOpts,
212 host = Host, user = User, password = Password, database = Database,
213 cap_found_rows = SetFoundRows} = State0) ->
214 %% Exchange handshake communication.
215 48 Result = mysql_protocol:handshake(Host, User, Password, Database, gen_tcp,
216 SSLOpts, Socket0, SetFoundRows),
217 48 case Result of
218 {ok, Handshake, SockMod, Socket} ->
219 46 setopts(SockMod, Socket, [{active, once}]),
220 46 #handshake{server_version = Version, connection_id = ConnId,
221 status = Status,
222 auth_plugin_name = AuthPluginName,
223 auth_plugin_data = AuthPluginData} = Handshake,
224 46 State1 = State0#state{server_version = Version, connection_id = ConnId,
225 sockmod = SockMod,
226 socket = Socket,
227 auth_plugin_name = AuthPluginName,
228 auth_plugin_data = AuthPluginData,
229 status = Status},
230 46 {ok, State1};
231 #error{} = E ->
232 2 {error, error_to_reason(E)}
233 end.
234
235 post_connect(#state{queries = Queries, prepares = Prepares} = State) ->
236 54 case execute_on_connect(Queries, Prepares, State) of
237 {ok, State1} ->
238 50 process_flag(trap_exit, true),
239 50 State2 = schedule_ping(State1),
240 50 {ok, State2};
241 {error, _} = E ->
242 4 E
243 end.
244
245 execute_on_connect([], [], State) ->
246 50 {ok, State};
247 execute_on_connect([], [{Name, Stmt}|Prepares], State) ->
248 7 case named_prepare(Name, Stmt, State) of
249 {{ok, Name}, State1} ->
250 5 execute_on_connect([], Prepares, State1);
251 {{error, _} = E, _} ->
252 2 E
253 end;
254 execute_on_connect([Query|Queries], Prepares, State) ->
255 15 case query(Query, no_filtermap_fun, default_timeout, State) of
256 {ok, State1} ->
257 5 execute_on_connect(Queries, Prepares, State1);
258 {{ok, _}, State1} ->
259 4 execute_on_connect(Queries, Prepares, State1);
260 {{ok, _, _}, State1} ->
261 4 execute_on_connect(Queries, Prepares, State1);
262 {{error, _} = E, _} ->
263 2 E
264 end.
265
266 %% @private
267 %% @doc
268 %%
269 %% Query and execute calls:
270 %%
271 %% <ul>
272 %% <li>{query, Query, FilterMap, Timeout}</li>
273 %% <li>{param_query, Query, Params, FilterMap, Timeout}</li>
274 %% <li>{execute, Stmt, Args, FilterMap, Timeout}</li>
275 %% </ul>
276 %%
277 %% For the calls listed above, we return these values:
278 %%
279 %% <dl>
280 %% <dt>`ok'</dt>
281 %% <dd>Success without returning any table data (UPDATE, etc.)</dd>
282 %% <dt>`{ok, ColumnNames, Rows}'</dt>
283 %% <dd>Queries returning one result set of table data</dd>
284 %% <dt>`{ok, [{ColumnNames, Rows}, ...]}'</dt>
285 %% <dd>Queries returning more than one result set of table data</dd>
286 %% <dt>`{error, ServerReason}'</dt>
287 %% <dd>MySQL server error</dd>
288 %% <dt>`{implicit_commit, NestingLevel, Query}'</dt>
289 %% <dd>A DDL statement (e.g. CREATE TABLE, ALTER TABLE, etc.) results in
290 %% an implicit commit.
291 %%
292 %% If the caller is in a (nested) transaction, it must be aborted. To be
293 %% able to handle this in the caller's process, we also return the
294 %% nesting level.</dd>
295 %% <dt>`{implicit_rollback, NestingLevel, ServerReason}'</dt>
296 %% <dd>This errors results in an implicit rollback: `{1213, <<"40001">>,
297 %% <<"Deadlock found when trying to get lock; try restarting "
298 %% "transaction">>}'.
299 %%
300 %% If the caller is in a (nested) transaction, it must be aborted. To be
301 %% able to handle this in the caller's process, we also return the
302 %% nesting level.</dd>
303 %% </dl>
304 handle_call(is_connected, _, #state{socket = Socket} = State) ->
305 4 {reply, Socket =/= undefined, State};
306 handle_call(Msg, From, #state{socket = undefined} = State) ->
307 1 case connect(State) of
308 {ok, State1} ->
309 1 handle_call(Msg, From, State1);
310 {error, _} = E ->
311 
:-(
 {stop, E, State}
312 end;
313 handle_call({query, _Query, _FilterMap, _Timeout}, {FromPid, _},
314 State = #state{transaction_levels = [{OtherFromPid, _} | _]})
315 when FromPid =/= OtherFromPid ->
316 %% this conn is currently in transaction owned by another process
317 
:-(
 {reply, {error, busy}, State};
318 handle_call({query, Query, FilterMap, Timeout}, _From, State) ->
319 480 {Reply, State1} = query(Query, FilterMap, Timeout, State),
320 480 {reply, Reply, State1};
321 handle_call({param_query, Query, Params, FilterMap, default_timeout}, From,
322 State) ->
323 16 handle_call({param_query, Query, Params, FilterMap,
324 State#state.query_timeout}, From, State);
325 handle_call({param_query, _Query, _Params, _FilterMap, _Timeout}, {FromPid, _},
326 State = #state{transaction_levels = [{OtherFromPid, _} | _]})
327 when FromPid =/= OtherFromPid ->
328 %% this conn is currently in transaction owned by another process
329 
:-(
 {reply, {error, busy}, State};
330 handle_call({param_query, Query, Params, FilterMap, Timeout}, _From,
331 #state{socket = Socket, sockmod = SockMod} = State) ->
332 %% Parametrized query: Prepared statement cached with the query as the key
333 17 QueryBin = iolist_to_binary(Query),
334 17 Cache = State#state.query_cache,
335 17 {StmtResult, Cache1} = case mysql_cache:lookup(QueryBin, Cache) of
336 {found, FoundStmt, NewCache} ->
337 %% Found
338 2 {{ok, FoundStmt}, NewCache};
339 not_found ->
340 %% Prepare
341 15 setopts(SockMod, Socket, [{active, false}]),
342 15 Rec = mysql_protocol:prepare(Query, SockMod, Socket),
343 15 setopts(SockMod, Socket, [{active, once}]),
344 15 case Rec of
345 #error{} = E ->
346 1 {{error, error_to_reason(E)}, Cache};
347 #prepared{} = Stmt ->
348 %% If the first entry in the cache, start the timer.
349 14 mysql_cache:is_empty(Cache) andalso begin
350 9 When = State#state.query_cache_time * 2,
351 9 erlang:send_after(When, self(), query_cache)
352 end,
353 14 {{ok, Stmt}, mysql_cache:store(QueryBin, Stmt, Cache)}
354 end
355 end,
356 17 case StmtResult of
357 {ok, StmtRec} ->
358 16 State1 = State#state{query_cache = Cache1},
359 16 {Reply, State2} = execute_stmt(StmtRec, Params, FilterMap, Timeout, State1),
360 16 {reply, Reply, State2};
361 PrepareError ->
362 1 {reply, PrepareError, State}
363 end;
364 handle_call({execute, Stmt, Args, FilterMap, default_timeout}, From, State) ->
365 207 handle_call({execute, Stmt, Args, FilterMap, State#state.query_timeout},
366 From, State);
367 handle_call({execute, _Stmt, _Args, _FilterMap, _Timeout}, {FromPid, _},
368 State = #state{transaction_levels = [{OtherFromPid, _} | _]})
369 when FromPid =/= OtherFromPid ->
370 %% this conn is currently in transaction owned by another process
371 
:-(
 {reply, {error, busy}, State};
372 handle_call({execute, Stmt, Args, FilterMap, Timeout}, _From, State) ->
373 208 case State#state.stmts of
374 #{Stmt := StmtRec} ->
375 207 {Reply, State1} = execute_stmt(StmtRec, Args, FilterMap, Timeout, State),
376 207 {reply, Reply, State1};
377 #{} ->
378 1 {reply, {error, not_prepared}, State}
379 end;
380 handle_call({prepare, Query}, _From, State) ->
381 95 #state{socket = Socket, sockmod = SockMod} = State,
382 95 setopts(SockMod, Socket, [{active, false}]),
383 95 Rec = mysql_protocol:prepare(Query, SockMod, Socket),
384 95 setopts(SockMod, Socket, [{active, once}]),
385 95 State1 = #state{stmts = Stmts1} = update_state(Rec, State),
386 95 case Rec of
387 #error{} = E ->
388 1 {reply, {error, error_to_reason(E)}, State1};
389 #prepared{statement_id = Id} = Stmt ->
390 94 State2 = State#state{stmts = Stmts1#{Id => Stmt}},
391 94 {reply, {ok, Id}, State2}
392 end;
393 handle_call({prepare, Name, Query}, _From, State) when is_atom(Name);
394 is_binary(Name) ->
395 9 {Reply, State1} = named_prepare(Name, Query, State),
396 9 {reply, Reply, State1};
397 handle_call({unprepare, Stmt}, _From, State) when is_atom(Stmt);
398 is_integer(Stmt);
399 is_binary(Stmt) ->
400 93 case maps:take(Stmt, State#state.stmts) of
401 {StmtRec, Stmts1} ->
402 91 #state{socket = Socket, sockmod = SockMod} = State,
403 91 setopts(SockMod, Socket, [{active, false}]),
404 91 mysql_protocol:unprepare(StmtRec, SockMod, Socket),
405 91 setopts(SockMod, Socket, [{active, once}]),
406 91 State1 = State#state{stmts = Stmts1},
407 91 State2 = schedule_ping(State1),
408 91 {reply, ok, State2};
409 error ->
410 2 {reply, {error, not_prepared}, State}
411 end;
412 handle_call({change_user, Username, Password, Options}, From,
413 State = #state{transaction_levels = []}) ->
414 9 #state{socket = Socket, sockmod = SockMod,
415 auth_plugin_name = AuthPluginName,
416 auth_plugin_data = AuthPluginData,
417 server_version = ServerVersion} = State,
418 9 Database = proplists:get_value(database, Options, undefined),
419 9 Queries = proplists:get_value(queries, Options, []),
420 9 Prepares = proplists:get_value(prepare, Options, []),
421 9 setopts(SockMod, Socket, [{active, false}]),
422 9 Result = mysql_protocol:change_user(SockMod, Socket, Username, Password,
423 AuthPluginName, AuthPluginData, Database,
424 ServerVersion),
425 9 setopts(SockMod, Socket, [{active, once}]),
426 9 State1 = update_state(Result, State),
427 9 State1#state.warning_count > 0 andalso State1#state.log_warnings
428 
:-(
 andalso log_warnings(State1, "CHANGE USER"),
429 9 State2 = State1#state{query_cache = mysql_cache:new(), stmts = #{}},
430 9 case Result of
431 #ok{} ->
432 8 State3 = State2#state{user = Username, password = Password,
433 database=Database, queries=Queries,
434 prepares=Prepares},
435 8 case post_connect(State3) of
436 {ok, State4} ->
437 6 {reply, ok, State4};
438 {error, Reason} = E ->
439 2 gen_server:reply(From, E),
440 2 stop_server(Reason, State3)
441 end;
442 #error{} = E ->
443 1 gen_server:reply(From, {error, error_to_reason(E)}),
444 1 stop_server(change_user_failed, State2)
445 end;
446 handle_call(reset_connection, _From, #state{socket = Socket, sockmod = SockMod} = State) ->
447 1 setopts(SockMod, Socket, [{active, false}]),
448 1 Result = mysql_protocol:reset_connnection(SockMod, Socket),
449 1 setopts(SockMod, Socket, [{active, once}]),
450 1 State1 = update_state(Result, State),
451 1 Reply = case Result of
452 1 #ok{} -> ok;
453 #error{} = E ->
454 %% 'COM_RESET_CONNECTION' is added in MySQL 5.7 and MariaDB 10
455 %% "Unknown command" is returned when MySQL =< 5.6 or MariaDB =< 5.5
456 
:-(
 {error, error_to_reason(E)}
457 end,
458 1 {reply, Reply, State1};
459
460 handle_call(warning_count, _From, State) ->
461 2 {reply, State#state.warning_count, State};
462 handle_call(insert_id, _From, State) ->
463 3 {reply, State#state.insert_id, State};
464 handle_call(affected_rows, _From, State) ->
465 3 {reply, State#state.affected_rows, State};
466 handle_call(autocommit, _From, State) ->
467 3 {reply, State#state.status band ?SERVER_STATUS_AUTOCOMMIT /= 0, State};
468 handle_call(backslash_escapes_enabled, _From, State = #state{status = S}) ->
469 2 {reply, S band ?SERVER_STATUS_NO_BACKSLASH_ESCAPES == 0, State};
470 handle_call(in_transaction, _From, State) ->
471 25 {reply, State#state.status band ?SERVER_STATUS_IN_TRANS /= 0, State};
472 handle_call(start_transaction, {FromPid, _},
473 State = #state{transaction_levels = [{OtherFromPid, _}|_]})
474 when FromPid =/= OtherFromPid ->
475 %% the idea of "nested transaction" is that mysql:transaction can be called
476 %% in the transaction function, but all within the same process
477 
:-(
 {reply, {error, busy}, State};
478 handle_call(start_transaction, {FromPid, _},
479 State = #state{socket = Socket, sockmod = SockMod,
480 decode_decimal = DecodeDecimal,
481 transaction_levels = L, status = Status})
482 when Status band ?SERVER_STATUS_IN_TRANS == 0, L == [];
483 Status band ?SERVER_STATUS_IN_TRANS /= 0, L /= [] ->
484 29 MRef = erlang:monitor(process, FromPid),
485 29 Query = case L of
486 18 [] -> <<"BEGIN">>;
487 11 _ -> <<"SAVEPOINT s", (integer_to_binary(length(L)))/binary>>
488 end,
489 29 setopts(SockMod, Socket, [{active, false}]),
490 29 {ok, [Res = #ok{}]} = mysql_protocol:query(Query, SockMod, Socket,
491 [], DecodeDecimal, no_filtermap_fun,
492 ?cmd_timeout),
493 29 setopts(SockMod, Socket, [{active, once}]),
494 29 State1 = update_state(Res, State),
495 29 {reply, ok, State1#state{transaction_levels = [{FromPid, MRef} | L]}};
496 handle_call(rollback, {FromPid, _},
497 State = #state{socket = Socket, sockmod = SockMod, status = Status,
498 decode_decimal = DecodeDecimal,
499 transaction_levels = [{FromPid, MRef} | L]})
500 when Status band ?SERVER_STATUS_IN_TRANS /= 0 ->
501 6 erlang:demonitor(MRef, [flush]),
502 6 Query = case L of
503 5 [] -> <<"ROLLBACK">>;
504 1 _ -> <<"ROLLBACK TO s", (integer_to_binary(length(L)))/binary>>
505 end,
506 6 setopts(SockMod, Socket, [{active, false}]),
507 6 {ok, [Res = #ok{}]} = mysql_protocol:query(Query, SockMod, Socket,
508 [], DecodeDecimal, no_filtermap_fun,
509 ?cmd_timeout),
510 6 setopts(SockMod, Socket, [{active, once}]),
511 6 State1 = update_state(Res, State),
512 6 {reply, ok, State1#state{transaction_levels = L}};
513 handle_call(commit, {FromPid, _},
514 State = #state{socket = Socket, sockmod = SockMod, status = Status,
515 decode_decimal = DecodeDecimal,
516 transaction_levels = [{FromPid, MRef} | L]})
517 when Status band ?SERVER_STATUS_IN_TRANS /= 0 ->
518 16 erlang:demonitor(MRef, [flush]),
519 16 Query = case L of
520 10 [] -> <<"COMMIT">>;
521 6 _ -> <<"RELEASE SAVEPOINT s", (integer_to_binary(length(L)))/binary>>
522 end,
523 16 setopts(SockMod, Socket, [{active, false}]),
524 16 {ok, [Res = #ok{}]} = mysql_protocol:query(Query, SockMod, Socket,
525 [], DecodeDecimal, no_filtermap_fun,
526 ?cmd_timeout),
527 16 setopts(SockMod, Socket, [{active, once}]),
528 16 State1 = update_state(Res, State),
529 16 {reply, ok, State1#state{transaction_levels = L}}.
530
531 %% @private
532 handle_cast(connect, #state{socket = undefined} = State) ->
533 2 case connect(State) of
534 {ok, State1} ->
535 1 {noreply, State1};
536 {error, _} = E ->
537 1 {stop, E, State}
538 end;
539 handle_cast(connect, State) ->
540 
:-(
 {noreply, State};
541 handle_cast(_Msg, State) ->
542 1 {noreply, State}.
543
544 %% @private
545 handle_info(query_cache, #state{query_cache = Cache,
546 query_cache_time = CacheTime} = State) ->
547 %% Evict expired queries/statements in the cache used by query/3.
548 1 {Evicted, Cache1} = mysql_cache:evict_older_than(Cache, CacheTime),
549 %% Unprepare the evicted statements
550 1 #state{socket = Socket, sockmod = SockMod} = State,
551 1 setopts(SockMod, Socket, [{active, false}]),
552 1 lists:foreach(fun ({_Query, Stmt}) ->
553 1 mysql_protocol:unprepare(Stmt, SockMod, Socket)
554 end,
555 Evicted),
556 1 setopts(SockMod, Socket, [{active, once}]),
557 %% If nonempty, schedule eviction again.
558 1 mysql_cache:is_empty(Cache1) orelse
559 
:-(
 erlang:send_after(CacheTime, self(), query_cache),
560 1 {noreply, State#state{query_cache = Cache1}};
561 handle_info({'DOWN', _MRef, _, Pid, _Info}, State) ->
562 2 stop_server({application_process_died, Pid}, State);
563 handle_info(ping, #state{socket = Socket, sockmod = SockMod} = State) ->
564 4 setopts(SockMod, Socket, [{active, false}]),
565 4 #ok{} = mysql_protocol:ping(SockMod, Socket),
566 3 setopts(SockMod, Socket, [{active, once}]),
567 3 {noreply, schedule_ping(State)};
568 handle_info({tcp_closed, Socket}, #state{socket = Socket} = State) ->
569 1 {stop, tcp_closed, State#state{socket = undefined, connection_id = undefined}};
570 handle_info({ssl_closed, Socket}, #state{socket = Socket} = State) ->
571 
:-(
 {stop, ssl_closed, State#state{socket = undefined, connection_id = undefined}};
572 handle_info({tcp_error, Socket, Reason}, #state{socket = Socket} = State) ->
573 1 stop_server({tcp_error, Reason}, State);
574 handle_info({ssl_error, Socket, Reason}, #state{socket = Socket} = State) ->
575 
:-(
 stop_server({ssl_error, Reason}, State);
576 handle_info({Transport, Socket, _Data}, #state{socket = Socket,
577 sockmod = SockMod} = State)
578 when Transport =:= tcp; Transport =:= ssl ->
579 %% Ignore out-of-band messages sent by the server,
580 %% eg. before closing the connection because of a session timeout
581 2 setopts(SockMod, Socket, [{active, once}]),
582 2 {noreply, State};
583 handle_info(_Info, State) ->
584 2 {noreply, State}.
585
586 %% @private
587 terminate(Reason, #state{socket = Socket, sockmod = SockMod})
588 when Socket =/= undefined andalso (Reason == normal orelse Reason == shutdown) ->
589 %% Send the goodbye message for politeness.
590 36 setopts(SockMod, Socket, [{active, false}]),
591 36 mysql_protocol:quit(SockMod, Socket);
592 terminate(_Reason, _State) ->
593 10 ok.
594
595 %% @private
596 code_change(_OldVsn, State = #state{}, _Extra) ->
597 2 {ok, State};
598 code_change(_OldVsn, _State, _Extra) ->
599 2 {error, incompatible_state}.
600
601 %% --- Helpers ---
602
603 %% @doc Executes a prepared statement and returns {Reply, NewState}.
604 execute_stmt(Stmt, Args, FilterMap, Timeout, State) ->
605 223 #state{socket = Socket, sockmod = SockMod,
606 allowed_local_paths = AllowedPaths,
607 float_as_decimal = FloatAsDecimal,
608 decode_decimal = DecodeDecimal} = State,
609 223 Args1 = case FloatAsDecimal of
610 false ->
611 216 Args;
612 _ ->
613 7 [float_to_decimal(Arg, FloatAsDecimal) || Arg <- Args]
614 end,
615 223 setopts(SockMod, Socket, [{active, false}]),
616 223 {ok, Recs} = case mysql_protocol:execute(Stmt, Args1, SockMod, Socket,
617 AllowedPaths, DecodeDecimal,
618 FilterMap, Timeout) of
619 {error, timeout} when State#state.server_version >= [5, 0, 0] ->
620 2 kill_query(State),
621 2 mysql_protocol:fetch_execute_response(SockMod, Socket, [],
622 DecodeDecimal, FilterMap,
623 ?cmd_timeout);
624 {error, Reason} ->
625 
:-(
 exit(Reason);
626 QueryResult ->
627 221 QueryResult
628 end,
629 223 setopts(SockMod, Socket, [{active, once}]),
630 223 State1 = lists:foldl(fun update_state/2, State, Recs),
631 223 State1#state.warning_count > 0 andalso State1#state.log_warnings
632 5 andalso log_warnings(State1, Stmt#prepared.orig_query),
633 223 handle_query_call_result(Recs, Stmt#prepared.orig_query, State1).
634
635 %% @doc Formats floats as decimals, optionally with a given number of decimals.
636 float_to_decimal(Arg, true) when is_float(Arg) ->
637 4 {decimal, list_to_binary(io_lib:format("~w", [Arg]))};
638 float_to_decimal(Arg, N) when is_float(Arg), is_integer(N) ->
639 1 {decimal, float_to_binary(Arg, [{decimals, N}, compact])};
640 float_to_decimal(Arg, _) ->
641 2 Arg.
642
643 %% @doc Produces a tuple to return as an error reason.
644 -spec error_to_reason(#error{}) -> mysql:server_reason().
645 error_to_reason(#error{code = Code, state = State, msg = Msg}) ->
646 19 {Code, State, Msg}.
647
648 %% @doc Updates a state with information from a response. Also re-schedules
649 %% ping.
650 -spec update_state(#ok{} | #eof{} | any(), #state{}) -> #state{}.
651 update_state(Rec, State) ->
652 908 State1 = case Rec of
653 #ok{status = S, affected_rows = R, insert_id = Id, warning_count = W} ->
654 488 State#state{status = S, affected_rows = R, insert_id = Id,
655 warning_count = W};
656 #resultset{status = S, warning_count = W} ->
657 297 State#state{status = S, warning_count = W};
658 #prepared{warning_count = W} ->
659 107 State#state{warning_count = W};
660 _Other ->
661 %% This includes errors.
662 %% Reset some things. (Note: We don't reset status and insert_id.)
663 16 State#state{warning_count = 0, affected_rows = 0}
664 end,
665 908 schedule_ping(State1).
666
667 %% @doc executes an unparameterized query and returns {Reply, NewState}.
668 query(Query, FilterMap, default_timeout,
669 #state{query_timeout = DefaultTimeout} = State) ->
670 493 query(Query, FilterMap, DefaultTimeout, State);
671 query(Query, FilterMap, Timeout, State) ->
672 495 #state{sockmod = SockMod, socket = Socket,
673 allowed_local_paths = AllowedPaths,
674 decode_decimal = DecodeDecimal} = State,
675 495 setopts(SockMod, Socket, [{active, false}]),
676 495 Result = mysql_protocol:query(Query, SockMod, Socket, AllowedPaths,
677 DecodeDecimal, FilterMap, Timeout),
678 495 {ok, Recs} = case Result of
679 {error, timeout} when State#state.server_version >= [5, 0, 0] ->
680 1 kill_query(State),
681 1 mysql_protocol:fetch_query_response(SockMod, Socket,
682 [], DecodeDecimal, FilterMap,
683 ?cmd_timeout);
684 {error, Reason} ->
685 
:-(
 exit(Reason);
686 QueryResult ->
687 494 QueryResult
688 end,
689 495 setopts(SockMod, Socket, [{active, once}]),
690 495 State1 = lists:foldl(fun update_state/2, State, Recs),
691 495 State1#state.warning_count > 0 andalso State1#state.log_warnings
692 3 andalso log_warnings(State1, Query),
693 495 handle_query_call_result(Recs, Query, State1).
694
695 %% @doc Prepares a named query and returns {{ok, Name}, NewState} or
696 %% {{error, Reason}, NewState}.
697 named_prepare(Name, Query, State) ->
698 16 #state{socket = Socket, sockmod = SockMod} = State,
699 %% First unprepare if there is an old statement with this name.
700 16 setopts(SockMod, Socket, [{active, false}]),
701 16 State1 = case maps:take(Name, State#state.stmts) of
702 {OldStmt, Stmts1} ->
703 1 mysql_protocol:unprepare(OldStmt, SockMod, Socket),
704 1 State#state{stmts = Stmts1};
705 error ->
706 15 State
707 end,
708 16 Rec = mysql_protocol:prepare(Query, SockMod, Socket),
709 16 setopts(SockMod, Socket, [{active, once}]),
710 16 State2 = #state{stmts = Stmts2} = update_state(Rec, State1),
711 16 case Rec of
712 #error{} = E ->
713 3 {{error, error_to_reason(E)}, State2};
714 #prepared{} = Stmt ->
715 13 State3 = State2#state{stmts = Stmts2#{Name => Stmt}},
716 13 {{ok, Name}, State3}
717 end.
718
719 %% @doc Transforms result sets into a structure appropriate to be returned
720 %% to the client.
721 handle_query_call_result([_] = Recs, Query, State) ->
722 706 handle_query_call_result(Recs, not_applicable, Query, State, []);
723 handle_query_call_result(Recs, Query, State) ->
724 12 handle_query_call_result(Recs, 1, Query, State, []).
725
726 handle_query_call_result([], _RecNum, _Query, State, []) ->
727 419 {ok, State};
728 handle_query_call_result([], _RecNum, _Query, State, [{ColumnNames, Rows}]) ->
729 279 {{ok, ColumnNames, Rows}, State};
730 handle_query_call_result([], _RecNum, _Query, State, ResultSetsAcc) ->
731 8 {{ok, lists:reverse(ResultSetsAcc)}, State};
732 handle_query_call_result([Rec|Recs], RecNum, Query,
733 #state{transaction_levels = L} = State,
734 ResultSetsAcc) ->
735 734 RecNum1 = case RecNum of
736 706 not_applicable -> not_applicable;
737 28 _ -> RecNum + 1
738 end,
739 734 case Rec of
740 #ok{status = Status} when Status band ?SERVER_STATUS_IN_TRANS == 0,
741 L /= [] ->
742 %% DDL statements (e.g. CREATE TABLE, ALTER TABLE, etc.) result in
743 %% an implicit commit.
744 1 Length = length(L),
745 1 Reply = {implicit_commit, Length, Query},
746 1 [] = demonitor_processes(L, Length),
747 1 {Reply, State#state{transaction_levels = []}};
748 #ok{status = Status} ->
749 425 maybe_log_slow_query(State, Status, RecNum, Query),
750 425 handle_query_call_result(Recs, RecNum1, Query, State, ResultSetsAcc);
751 #resultset{cols = ColDefs, rows = Rows, status = Status} ->
752 297 Names = [Def#col.name || Def <- ColDefs],
753 297 ResultSetsAcc1 = [{Names, Rows} | ResultSetsAcc],
754 297 maybe_log_slow_query(State, Status, RecNum, Query),
755 297 handle_query_call_result(Recs, RecNum1, Query, State, ResultSetsAcc1);
756 #error{code = ?ERROR_DEADLOCK} when L /= [] ->
757 %% These errors result in an implicit rollback.
758 3 Reply = {implicit_rollback, length(L), error_to_reason(Rec)},
759 %% The transaction is rollbacked, except the BEGIN, so we're still
760 %% in a transaction. (In 5.7+, also the BEGIN has been rolled back,
761 %% but here we assume the old behaviour.)
762 3 NewMonitors = demonitor_processes(L, length(L) - 1),
763 3 {Reply, State#state{transaction_levels = NewMonitors}};
764 #error{} ->
765 8 {{error, error_to_reason(Rec)}, State}
766 end.
767
768 %% @doc Schedules (or re-schedules) ping.
769 schedule_ping(State = #state{ping_timeout = infinity}) ->
770 301 State;
771 schedule_ping(State = #state{ping_timeout = Timeout, ping_ref = Ref}) ->
772 751 is_reference(Ref) andalso erlang:cancel_timer(Ref),
773 751 State#state{ping_ref = erlang:send_after(Timeout, self(), ping)}.
774
775 %% @doc Fetches and logs warnings. Query is the query that gave the warnings.
776 log_warnings(#state{socket = Socket, sockmod = SockMod,
777 decode_decimal = DecodeDecimal}, Query) ->
778 8 setopts(SockMod, Socket, [{active, false}]),
779 8 {ok, [#resultset{rows = Rows}]} = mysql_protocol:query(<<"SHOW WARNINGS">>,
780 SockMod, Socket,
781 [], DecodeDecimal,
782 no_filtermap_fun,
783 ?cmd_timeout),
784 8 setopts(SockMod, Socket, [{active, once}]),
785 8 Lines = [[Level, " ", integer_to_binary(Code), ": ", Message, "\n"]
786 8 || [Level, Code, Message] <- Rows],
787 8 logger:warning("~s in ~s~n", [Lines, Query]).
788
789 %% @doc Logs slow queries. Query is the query that gave the warnings.
790 maybe_log_slow_query(#state{log_slow_queries = true}, S, RecNum, Query)
791 when S band ?SERVER_QUERY_WAS_SLOW /= 0 ->
792 3 IndexHint = if
793 S band ?SERVER_STATUS_NO_GOOD_INDEX_USED /= 0 ->
794 
:-(
 " (with no good index)";
795 S band ?SERVER_STATUS_NO_INDEX_USED /= 0 ->
796 
:-(
 " (with no index)";
797 true ->
798 3 ""
799 end,
800 3 QueryNumHint = case RecNum of
801 not_applicable ->
802 1 "";
803 _ ->
804 2 io_lib:format(" #~b", [RecNum])
805 end,
806 3 logger:warning("MySQL query~s~s was slow: ~s~n",
807 [QueryNumHint, IndexHint, Query]);
808 maybe_log_slow_query(_, _, _, _) ->
809 719 ok.
810
811 %% @doc Makes a separate connection and execute KILL QUERY. We do this to get
812 %% our main connection back to normal. KILL QUERY appeared in MySQL 5.0.0.
813 kill_query(#state{connection_id = ConnId, host = Host, port = Port,
814 user = User, password = Password, ssl_opts = SSLOpts,
815 cap_found_rows = SetFoundRows}) ->
816 %% Connect socket
817 3 SockOpts = [{active, false}, binary, {packet, raw}],
818 3 {ok, Socket0} = gen_tcp:connect(Host, Port, SockOpts),
819
820 %% Exchange handshake communication.
821 3 Result = mysql_protocol:handshake(Host, User, Password, undefined, gen_tcp,
822 SSLOpts, Socket0, SetFoundRows),
823 3 case Result of
824 {ok, #handshake{}, SockMod, Socket} ->
825 %% Kill and disconnect
826 3 IdBin = integer_to_binary(ConnId),
827 3 {ok, [#ok{}]} = mysql_protocol:query(<<"KILL QUERY ", IdBin/binary>>,
828 SockMod, Socket,
829 [], auto,
830 no_filtermap_fun,
831 ?cmd_timeout),
832 3 mysql_protocol:quit(SockMod, Socket);
833 #error{} = E ->
834 
:-(
 logger:error("Failed to connect to kill query: ~p",
835 [error_to_reason(E)])
836 end.
837
838 stop_server(Reason, #state{socket = undefined} = State) ->
839 
:-(
 {stop, Reason, State};
840 stop_server(Reason,
841 #state{socket = Socket, connection_id = ConnId, sockmod = SockMod} = State) ->
842 6 logger:error("Connection Id ~p closing with reason: ~p~n",
843 [ConnId, Reason]),
844 6 ok = SockMod:close(Socket),
845 6 {stop, Reason, State#state{socket = undefined, connection_id = undefined}}.
846
847 setopts(gen_tcp, Socket, Opts) ->
848 2078 inet:setopts(Socket, Opts);
849 setopts(SockMod, Socket, Opts) ->
850 23 SockMod:setopts(Socket, Opts).
851
852 demonitor_processes(List, 0) ->
853 4 List;
854 demonitor_processes([{_FromPid, MRef}|T], Count) ->
855 5 erlang:demonitor(MRef, [flush]),
856 5 demonitor_processes(T, Count - 1).
857
858 format_status(Status = #{state := State = #state{ssl_opts = SSLOpts}}) ->
859 12 Status#{state := State#state{password = hidden,
860 ssl_opts = obfuscate_ssl_options(SSLOpts)}}.
861
862 -if(?OTP_RELEASE < 25).
863 format_status(_, [_PDict, State = #state{ssl_opts = SSLOpts}]) ->
864 {data, [{"State", State#state{password = hidden,
865 ssl_opts = obfuscate_ssl_options(SSLOpts)}}]}.
866 -endif.
867
868 obfuscate_ssl_options(undefined) ->
869 11 undefined;
870 obfuscate_ssl_options(SSLOpts) ->
871 1 lists:map(
872 fun
873 
:-(
 ({cert, _}) -> {cert, hidden};
874 
:-(
 ({key, _}) -> {key, hidden};
875 
:-(
 ({cacerts, _}) -> {cacerts, hidden};
876 2 (Other) -> Other
877 end,
878 SSLOpts
879 ).
Line Hits Source