agora inbox for pgsql-hackers@postgresql.orghelp / color / mirror / Atom feed
[PATCH v45 2/3] Filling gaps in jsonb 8+ messages / 2 participants [nested] [flat]
* [PATCH v45 2/3] Filling gaps in jsonb @ 2020-12-31 14:19 Dmitrii Dolgov <9erthalion6@gmail.com> 0 siblings, 0 replies; 8+ messages in thread From: Dmitrii Dolgov @ 2020-12-31 14:19 UTC (permalink / raw) Introduces two new modes for jsonb assignment: * Appending array elements on the specified position, gaps filled with nulls (similar to JavaScript behavior). This mode also instructs to create the whole path in a jsonb object if some part of the path (more than just the last element) is not present. * Assigning keeps array positions consistent by prevent prepending of elements. Originally proposed by Nikita Glukhov based on polymorphic subscripting patch, but transformed into an independent change. --- doc/src/sgml/json.sgml | 33 ++++ src/backend/utils/adt/jsonfuncs.c | 226 ++++++++++++++++++++++++++-- src/test/regress/expected/jsonb.out | 135 +++++++++++++++++ src/test/regress/sql/jsonb.sql | 81 ++++++++++ 4 files changed, 460 insertions(+), 15 deletions(-) diff --git a/doc/src/sgml/json.sgml b/doc/src/sgml/json.sgml index 100d1a60f4..9af015d222 100644 --- a/doc/src/sgml/json.sgml +++ b/doc/src/sgml/json.sgml @@ -645,6 +645,39 @@ UPDATE table_name SET jsonb_field['a'] = '1'; -- If jsonb_field here is NULL, the result is [1] UPDATE table_name SET jsonb_field[0] = '1'; +</programlisting> + + Jsonb assignment via subscripting handles few edge cases differently + from <literal>jsonb_set</literal>. When assigning to the jsonb array + to the specified index, but there are no other elements present, the + result will be a jsonb array with the ewn value by specified index and + <type>null</type> elements from the first index to the specified index. + +<programlisting> +-- If jsonb_field is [], the result is [null, null, 2] +UPDATE table_name SET jsonb_field[2] = '2'; +</programlisting> + + When assigning to the jsonb array to the specified index, but position + of the last element in the array is less than the specified index, the + result will be a jsonb array with the new value by specified index and + <type>null</type> elements from the last index to the specified index. + +<programlisting> +-- If jsonb_field is [0], the result is [0, null, 2] +UPDATE table_name SET jsonb_field[2] = '2'; +</programlisting> + + When assigning using the path which is not present in the source jsonb, + the result will be a jsonb with the specified path created and the new + value at the end of the path. + +<programlisting> +-- If jsonb_field is {}, the result is {'a': [{'b': 1}]} +UPDATE table_name SET jsonb_field['a'][0]['b'] = '1'; + +-- If jsonb_field is [], the result is [{'a': 1}] +UPDATE table_name SET jsonb_field[0]['a'] = '1'; </programlisting> </para> diff --git a/src/backend/utils/adt/jsonfuncs.c b/src/backend/utils/adt/jsonfuncs.c index 5a0ba6b220..f14f6c3191 100644 --- a/src/backend/utils/adt/jsonfuncs.c +++ b/src/backend/utils/adt/jsonfuncs.c @@ -44,6 +44,8 @@ #define JB_PATH_INSERT_AFTER 0x0010 #define JB_PATH_CREATE_OR_INSERT \ (JB_PATH_INSERT_BEFORE | JB_PATH_INSERT_AFTER | JB_PATH_CREATE) +#define JB_PATH_FILL_GAPS 0x0020 +#define JB_PATH_CONSISTENT_POSITION 0x0040 /* state for json_object_keys */ typedef struct OkeysState @@ -1634,14 +1636,116 @@ jsonb_set_element(Jsonb* jb, Datum *path, int path_len, it = JsonbIteratorInit(&jb->root); - res = setPath(&it, path, path_nulls, path_len, &state, 0, - newval, JB_PATH_CREATE); + res = setPath(&it, path, path_nulls, path_len, &state, 0, newval, + JB_PATH_CREATE | JB_PATH_FILL_GAPS | + JB_PATH_CONSISTENT_POSITION); pfree(path_nulls); PG_RETURN_JSONB_P(JsonbValueToJsonb(res)); } +static void +push_null_elements(JsonbParseState **ps, int num) +{ + JsonbValue null; + + null.type = jbvNull; + + while (num-- > 0) + pushJsonbValue(ps, WJB_ELEM, &null); +} + +/* + * Prepare a new structure containing nested empty objects and arrays + * corresponding to the specified path, and assign a new value at the end of + * this path. E.g. the path [a][0][b] with the new value 1 will produce the + * structure {a: [{b: 1}]}. + * + * Called is responsible to make sure such path does not exist yet. + */ +static void +push_path(JsonbParseState **st, int level, Datum *path_elems, + bool *path_nulls, int path_len, JsonbValue *newval) +{ + /* + * tpath contains expected type of an empty jsonb created at each level + * higher or equal than the current one, either jbvObject or jbvArray. + * Since it contains only information about path slice from level to the + * end, the access index must be normalized by level. + */ + enum jbvType *tpath = palloc0((path_len - level) * sizeof(enum jbvType)); + long lindex; + JsonbValue newkey; + + /* + * Create first part of the chain with beginning tokens. For the current + * level WJB_BEGIN_OBJECT/WJB_BEGIN_ARRAY was already created, so start + * with the next one. + */ + for(int i = level + 1; i < path_len; i++) + { + char *c, *badp; + + if (path_nulls[i]) + break; + + /* + * Try to convert to an integer to find out the expected type, + * object or array. + */ + c = TextDatumGetCString(path_elems[i]); + errno = 0; + lindex = strtol(c, &badp, 10); + if (errno != 0 || badp == c || *badp != '\0' || lindex > INT_MAX || + lindex < INT_MIN) + { + /* text, an object is expected */ + newkey.type = jbvString; + newkey.val.string.len = VARSIZE_ANY_EXHDR(path_elems[i]); + newkey.val.string.val = VARDATA_ANY(path_elems[i]); + + (void) pushJsonbValue(st, WJB_BEGIN_OBJECT, NULL); + (void) pushJsonbValue(st, WJB_KEY, &newkey); + + tpath[i - level] = jbvObject; + } + else + { + /* integer, an array is expected */ + (void) pushJsonbValue(st, WJB_BEGIN_ARRAY, NULL); + + push_null_elements(st, lindex); + + tpath[i - level] = jbvArray; + } + + } + + /* Insert an actual value for either an object or array */ + if (tpath[(path_len - level) - 1] == jbvArray) + { + (void) pushJsonbValue(st, WJB_ELEM, newval); + } + else + (void) pushJsonbValue(st, WJB_VALUE, newval); + + /* + * Close everything up to the last but one level. The last one will be + * closed outside of this function. + */ + for(int i = path_len - 1; i > level; i--) + { + if (path_nulls[i]) + break; + + if (tpath[i - level] == jbvObject) + (void) pushJsonbValue(st, WJB_END_OBJECT, NULL); + else + (void) pushJsonbValue(st, WJB_END_ARRAY, NULL); + } +} + /* * Return the text representation of the given JsonbValue. */ @@ -4782,6 +4886,21 @@ IteratorConcat(JsonbIterator **it1, JsonbIterator **it2, * Bits JB_PATH_INSERT_BEFORE and JB_PATH_INSERT_AFTER in op_type * behave as JB_PATH_CREATE if new value is inserted in JsonbObject. * + * If JB_PATH_FILL_GAPS bit is set, this will change an assignment logic in + * case if target is an array. The assignment index will not be restricted by + * number of elements in the array, and if there are any empty slots between + * last element of the array and a new one they will be filled with nulls. If + * the index is negative, it still will be considered an an index from the end + * of the array. Of a part of the path is not present and this part is more + * than just one last element, this flag will instruct to create the whole + * chain of corresponding objects and insert the value. + * + * JB_PATH_CONSISTENT_POSITION for an array indicates that the called wants to + * keep values with fixed indices. Indices for existing elements could be + * changed (shifted forward) in case if the array is prepended with a new value + * and a negative index out of the range, so this behavior will be prevented + * and return an error. + * * All path elements before the last must already exist * whatever bits in op_type are set, or nothing is done. */ @@ -4876,6 +4995,8 @@ setPathObject(JsonbIterator **it, Datum *path_elems, bool *path_nulls, memcmp(k.val.string.val, VARDATA_ANY(path_elems[level]), k.val.string.len) == 0) { + done = true; + if (level == path_len - 1) { /* @@ -4895,7 +5016,6 @@ setPathObject(JsonbIterator **it, Datum *path_elems, bool *path_nulls, (void) pushJsonbValue(st, WJB_KEY, &k); (void) pushJsonbValue(st, WJB_VALUE, newval); } - done = true; } else { @@ -4940,6 +5060,31 @@ setPathObject(JsonbIterator **it, Datum *path_elems, bool *path_nulls, } } } + + /* + * If we got here there are only few possibilities: + * - no target path was found, and an open object with some keys/values was + * pushed into the state + * - an object is empty, only WJB_BEGIN_OBJECT is pushed + * + * In both cases if instructed to create the path when not present, + * generate the whole chain of empty objects and insert the new value + * there. + */ + if (!done && (op_type & JB_PATH_FILL_GAPS) && (level < path_len - 1)) + { + JsonbValue newkey; + + newkey.type = jbvString; + newkey.val.string.len = VARSIZE_ANY_EXHDR(path_elems[level]); + newkey.val.string.val = VARDATA_ANY(path_elems[level]); + + (void) pushJsonbValue(st, WJB_KEY, &newkey); + (void) push_path(st, level, path_elems, path_nulls, + path_len, newval); + + /* Result is closed with WJB_END_OBJECT outside of this function */ + } } /* @@ -4978,25 +5123,48 @@ setPathArray(JsonbIterator **it, Datum *path_elems, bool *path_nulls, if (idx < 0) { if (-idx > nelems) - idx = INT_MIN; + { + /* + * If asked to keep elements position consistent, it's not allowed + * to prepend the array. + */ + if (op_type & JB_PATH_CONSISTENT_POSITION) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("path element at position %d is out of range: %d", + level + 1, idx))); + else + idx = INT_MIN; + } else idx = nelems + idx; } - if (idx > 0 && idx > nelems) - idx = nelems; + /* + * Filling the gaps means there are no limits on the positive index are + * imposed, we can set any element. Otherwise limit the index by nelems. + */ + if (!(op_type & JB_PATH_FILL_GAPS)) + { + if (idx > 0 && idx > nelems) + idx = nelems; + } /* * if we're creating, and idx == INT_MIN, we prepend the new value to the * array also if the array is empty - in which case we don't really care * what the idx value is */ - if ((idx == INT_MIN || nelems == 0) && (level == path_len - 1) && (op_type & JB_PATH_CREATE_OR_INSERT)) { Assert(newval != NULL); + + if (op_type & JB_PATH_FILL_GAPS && nelems == 0 && idx > 0) + push_null_elements(st, idx); + (void) pushJsonbValue(st, WJB_ELEM, newval); + done = true; } @@ -5007,6 +5175,8 @@ setPathArray(JsonbIterator **it, Datum *path_elems, bool *path_nulls, if (i == idx && level < path_len) { + done = true; + if (level == path_len - 1) { r = JsonbIteratorNext(it, &v, true); /* skip */ @@ -5024,8 +5194,6 @@ setPathArray(JsonbIterator **it, Datum *path_elems, bool *path_nulls, if (op_type & (JB_PATH_INSERT_AFTER | JB_PATH_REPLACE)) (void) pushJsonbValue(st, WJB_ELEM, newval); - - done = true; } else (void) setPath(it, path_elems, path_nulls, path_len, @@ -5053,14 +5221,42 @@ setPathArray(JsonbIterator **it, Datum *path_elems, bool *path_nulls, (void) pushJsonbValue(st, r, r < WJB_BEGIN_ARRAY ? &v : NULL); } } - - if ((op_type & JB_PATH_CREATE_OR_INSERT) && !done && - level == path_len - 1 && i == nelems - 1) - { - (void) pushJsonbValue(st, WJB_ELEM, newval); - } } } + + if ((op_type & JB_PATH_CREATE_OR_INSERT) && !done && level == path_len - 1) + { + /* + * If asked to fill the gaps, idx could be bigger than nelems, + * so prepend the new element with nulls if that's the case. + */ + if (op_type & JB_PATH_FILL_GAPS && idx > nelems) + push_null_elements(st, idx - nelems); + + (void) pushJsonbValue(st, WJB_ELEM, newval); + done = true; + } + + /* + * If we got here there are only few possibilities: + * - no target path was found, and an open array with some keys/values was + * pushed into the state + * - an array is empty, only WJB_BEGIN_ARRAY is pushed + * + * In both cases if instructed to create the path when not present, + * generate the whole chain of empty objects and insert the new value + * there. + */ + if (!done && (op_type & JB_PATH_FILL_GAPS) && (level < path_len - 1)) + { + if (idx > 0) + push_null_elements(st, idx - nelems); + + (void) push_path(st, level, path_elems, path_nulls, + path_len, newval); + + /* Result is closed with WJB_END_OBJECT outside of this function */ + } } /* diff --git a/src/test/regress/expected/jsonb.out b/src/test/regress/expected/jsonb.out index bb3f25ec3f..b7c268b53f 100644 --- a/src/test/regress/expected/jsonb.out +++ b/src/test/regress/expected/jsonb.out @@ -4999,6 +4999,141 @@ select * from test_jsonb_subscript; 3 | [1] (3 rows) +-- Fill the gaps logic +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[0]'); +update test_jsonb_subscript set test_json[5] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+-------------------------------- + 1 | [0, null, null, null, null, 1] +(1 row) + +update test_jsonb_subscript set test_json[-4] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+----------------------------- + 1 | [0, null, 1, null, null, 1] +(1 row) + +update test_jsonb_subscript set test_json[-8] = '1'; +ERROR: path element at position 1 is out of range: -8 +select * from test_jsonb_subscript; + id | test_json +----+----------------------------- + 1 | [0, null, 1, null, null, 1] +(1 row) + +-- keep consistent values position +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[]'); +update test_jsonb_subscript set test_json[5] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+----------------------------------- + 1 | [null, null, null, null, null, 1] +(1 row) + +-- create the whole path +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a'][0]['b'][0]['c'] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+---------------------------- + 1 | {"a": [{"b": [{"c": 1}]}]} +(1 row) + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a'][2]['b'][2]['c'][2] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+------------------------------------------------------------------ + 1 | {"a": [null, null, {"b": [null, null, {"c": [null, null, 1]}]}]} +(1 row) + +-- create the whole path with already existing keys +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{"b": 1}'); +update test_jsonb_subscript set test_json['a'][0] = '2'; +select * from test_jsonb_subscript; + id | test_json +----+-------------------- + 1 | {"a": [2], "b": 1} +(1 row) + +-- the start jsonb is an object, first subscript is treated as a key +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json[0]['a'] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+----------------- + 1 | {"0": {"a": 1}} +(1 row) + +-- the start jsonb is an array +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[]'); +update test_jsonb_subscript set test_json[0]['a'] = '1'; +update test_jsonb_subscript set test_json[2]['b'] = '2'; +select * from test_jsonb_subscript; + id | test_json +----+---------------------------- + 1 | [{"a": 1}, null, {"b": 2}] +(1 row) + +-- overwriting an existing path +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a']['b'][1] = '1'; +update test_jsonb_subscript set test_json['a']['b'][10] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+---------------------------------------------------------------------------- + 1 | {"a": {"b": [null, 1, null, null, null, null, null, null, null, null, 1]}} +(1 row) + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[]'); +update test_jsonb_subscript set test_json[0][0][0] = '1'; +update test_jsonb_subscript set test_json[0][0][1] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+------------ + 1 | [[[1, 1]]] +(1 row) + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a']['b'][10] = '1'; +update test_jsonb_subscript set test_json['a'][10][10] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+------------------------------------------------------------------------------------------------------------------------------------------------------ + 1 | {"a": {"b": [null, null, null, null, null, null, null, null, null, null, 1], "10": [null, null, null, null, null, null, null, null, null, null, 1]}} +(1 row) + +-- an empty sub element +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{"a": {}}'); +update test_jsonb_subscript set test_json['a']['b']['c'][2] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+-------------------------------------- + 1 | {"a": {"b": {"c": [null, null, 1]}}} +(1 row) + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{"a": []}'); +update test_jsonb_subscript set test_json['a'][1]['c'][2] = '1'; +select * from test_jsonb_subscript; + id | test_json +----+--------------------------------------- + 1 | {"a": [null, {"c": [null, null, 1]}]} +(1 row) + -- jsonb to tsvector select to_tsvector('{"a": "aaa bbb ddd ccc", "b": ["eee fff ggg"], "c": {"d": "hhh iii"}}'::jsonb); to_tsvector diff --git a/src/test/regress/sql/jsonb.sql b/src/test/regress/sql/jsonb.sql index 20aa8fe0e2..0320db0ea4 100644 --- a/src/test/regress/sql/jsonb.sql +++ b/src/test/regress/sql/jsonb.sql @@ -1290,6 +1290,87 @@ update test_jsonb_subscript set test_json = NULL where id = 3; update test_jsonb_subscript set test_json[0] = '1'; select * from test_jsonb_subscript; +-- Fill the gaps logic +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[0]'); + +update test_jsonb_subscript set test_json[5] = '1'; +select * from test_jsonb_subscript; + +update test_jsonb_subscript set test_json[-4] = '1'; +select * from test_jsonb_subscript; + +update test_jsonb_subscript set test_json[-8] = '1'; +select * from test_jsonb_subscript; + +-- keep consistent values position +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[]'); + +update test_jsonb_subscript set test_json[5] = '1'; +select * from test_jsonb_subscript; + +-- create the whole path +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a'][0]['b'][0]['c'] = '1'; +select * from test_jsonb_subscript; + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a'][2]['b'][2]['c'][2] = '1'; +select * from test_jsonb_subscript; + +-- create the whole path with already existing keys +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{"b": 1}'); +update test_jsonb_subscript set test_json['a'][0] = '2'; +select * from test_jsonb_subscript; + +-- the start jsonb is an object, first subscript is treated as a key +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json[0]['a'] = '1'; +select * from test_jsonb_subscript; + +-- the start jsonb is an array +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[]'); +update test_jsonb_subscript set test_json[0]['a'] = '1'; +update test_jsonb_subscript set test_json[2]['b'] = '2'; +select * from test_jsonb_subscript; + +-- overwriting an existing path +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a']['b'][1] = '1'; +update test_jsonb_subscript set test_json['a']['b'][10] = '1'; +select * from test_jsonb_subscript; + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '[]'); +update test_jsonb_subscript set test_json[0][0][0] = '1'; +update test_jsonb_subscript set test_json[0][0][1] = '1'; +select * from test_jsonb_subscript; + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{}'); +update test_jsonb_subscript set test_json['a']['b'][10] = '1'; +update test_jsonb_subscript set test_json['a'][10][10] = '1'; +select * from test_jsonb_subscript; + +-- an empty sub element + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{"a": {}}'); +update test_jsonb_subscript set test_json['a']['b']['c'][2] = '1'; +select * from test_jsonb_subscript; + +delete from test_jsonb_subscript; +insert into test_jsonb_subscript values (1, '{"a": []}'); +update test_jsonb_subscript set test_json['a'][1]['c'][2] = '1'; +select * from test_jsonb_subscript; + -- jsonb to tsvector select to_tsvector('{"a": "aaa bbb ddd ccc", "b": ["eee fff ggg"], "c": {"d": "hhh iii"}}'::jsonb); -- 2.21.0 --2v7pubbjx3p2om6u Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v45-0003-Replace-assuming-a-composite-object-on-a-scalar.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.8 26/38] aio: Basic read_stream adjustments for real AIO @ 2025-03-14 15:39 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-14 15:39 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency There is one comment talking about max_ios logic with "real asynchronous I/O" that I am not sure about, so I left it alone for now. There are further improvements we should consider, e.g. waiting to issue IOs until we can issue multiple IOs at once. But that's left for a future change, since it would involve additional heuristics. --- src/backend/storage/aio/read_stream.c | 39 ++++++++++++++++++++------- 1 file changed, 29 insertions(+), 10 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index 0f1525f46c5..b44cc358f29 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,7 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + bool sync_mode; bool advice_enabled; bool temporary; @@ -420,6 +422,13 @@ read_stream_start_pending_read(ReadStream *stream) static void read_stream_look_ahead(ReadStream *stream) { + /* + * Allow amortizing the cost of submitting IO over multiple IOs. This + * requires that we don't do any operations that could lead to a deadlock + * with staged-but-unsubmitted IO. + */ + pgaio_enter_batchmode(); + while (stream->ios_in_progress < stream->max_ios && stream->pinned_buffers + stream->pending_read_nblocks < stream->distance) { @@ -467,6 +476,7 @@ read_stream_look_ahead(ReadStream *stream) { /* We've hit the buffer or I/O limit. Rewind and stop here. */ read_stream_unget_block(stream, blocknum); + pgaio_exit_batchmode(); return; } } @@ -501,6 +511,8 @@ read_stream_look_ahead(ReadStream *stream) * time. */ Assert(stream->pinned_buffers > 0 || stream->distance == 0); + + pgaio_exit_batchmode(); } /* @@ -556,12 +568,12 @@ read_stream_begin_impl(int flags, max_ios = get_tablespace_io_concurrency(tablespace_id); /* - * XXX Since we don't have asynchronous I/O yet, if direct I/O is enabled - * then just behave as though I/O concurrency is set to 0. Otherwise we - * would look ahead pinning many buffers for no benefit, for lack of - * advice and AIO. + * If real asynchronous I/O is disabled, and direct I/O is enabled, just + * behave as though I/O concurrency is set to 0. Otherwise we would look + * ahead pinning many buffers for no benefit, as the advice-based + * readahead doesn't support direct I/O. */ - if (io_direct_flags & IO_DIRECT_DATA) + if (io_method == IOMETHOD_SYNC && (io_direct_flags & IO_DIRECT_DATA)) max_ios = 0; /* Cap to INT16_MAX to avoid overflowing below */ @@ -641,15 +653,19 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * This system supports prefetching advice. + * + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; @@ -659,6 +675,9 @@ read_stream_begin_impl(int flags, * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled * above. If we had real asynchronous I/O we might need a slightly * different definition. + * + * FIXME: Not sure what different definition we would need? I guess we + * could add the READ_BUFFERS_SYNCHRONOUSLY flag automatically? */ if (max_ios == 0) max_ios = 1; -- 2.48.1.76.g4e746b1a31.dirty --ow5flh3n247znjrs Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.8-0027-aio-Experimental-heuristics-to-increase-batchin.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.9 18/30] aio: Basic read_stream adjustments for real AIO @ 2025-03-15 16:29 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-15 16:29 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency There is one comment talking about max_ios logic with "real asynchronous I/O" that I am not sure about, so I left it alone for now. There are further improvements we should consider, e.g. waiting to issue IOs until we can issue multiple IOs at once. But that's left for a future change, since it would involve additional heuristics. --- src/backend/storage/aio/read_stream.c | 29 ++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index cdf4b5a86a2..d7b395b86a3 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,7 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + bool sync_mode; bool advice_enabled; bool temporary; @@ -416,6 +418,13 @@ read_stream_start_pending_read(ReadStream *stream) static void read_stream_look_ahead(ReadStream *stream) { + /* + * Allow amortizing the cost of submitting IO over multiple IOs. This + * requires that we don't do any operations that could lead to a deadlock + * with staged-but-unsubmitted IO. + */ + pgaio_enter_batchmode(); + while (stream->ios_in_progress < stream->max_ios && stream->pinned_buffers + stream->pending_read_nblocks < stream->distance) { @@ -463,6 +472,7 @@ read_stream_look_ahead(ReadStream *stream) { /* We've hit the buffer or I/O limit. Rewind and stop here. */ read_stream_unget_block(stream, blocknum); + pgaio_exit_batchmode(); return; } } @@ -497,6 +507,8 @@ read_stream_look_ahead(ReadStream *stream) * time. */ Assert(stream->pinned_buffers > 0 || stream->distance == 0); + + pgaio_exit_batchmode(); } /* @@ -628,15 +640,19 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * This system supports prefetching advice. + * + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; @@ -646,6 +662,9 @@ read_stream_begin_impl(int flags, * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled * above. If we had real asynchronous I/O we might need a slightly * different definition. + * + * FIXME: Not sure what different definition we would need? I guess we + * could add the READ_BUFFERS_SYNCHRONOUSLY flag automatically? */ if (max_ios == 0) max_ios = 1; -- 2.48.1.76.g4e746b1a31.dirty --23jbdfobqrqxnmx5 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.9-0019-aio-Experimental-heuristics-to-increase-batchin.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.10 12/28] aio: Basic read_stream adjustments for real AIO @ 2025-03-18 18:40 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency There is one comment talking about max_ios logic with "real asynchronous I/O" that I am not sure about, so I left it alone for now. There are further improvements we should consider, e.g. waiting to issue IOs until we can issue multiple IOs at once. But that's left for a future change, since it would involve additional heuristics. --- src/backend/storage/aio/read_stream.c | 29 ++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index cdf4b5a86a2..d7b395b86a3 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,7 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + bool sync_mode; bool advice_enabled; bool temporary; @@ -416,6 +418,13 @@ read_stream_start_pending_read(ReadStream *stream) static void read_stream_look_ahead(ReadStream *stream) { + /* + * Allow amortizing the cost of submitting IO over multiple IOs. This + * requires that we don't do any operations that could lead to a deadlock + * with staged-but-unsubmitted IO. + */ + pgaio_enter_batchmode(); + while (stream->ios_in_progress < stream->max_ios && stream->pinned_buffers + stream->pending_read_nblocks < stream->distance) { @@ -463,6 +472,7 @@ read_stream_look_ahead(ReadStream *stream) { /* We've hit the buffer or I/O limit. Rewind and stop here. */ read_stream_unget_block(stream, blocknum); + pgaio_exit_batchmode(); return; } } @@ -497,6 +507,8 @@ read_stream_look_ahead(ReadStream *stream) * time. */ Assert(stream->pinned_buffers > 0 || stream->distance == 0); + + pgaio_exit_batchmode(); } /* @@ -628,15 +640,19 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * This system supports prefetching advice. + * + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; @@ -646,6 +662,9 @@ read_stream_begin_impl(int flags, * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled * above. If we had real asynchronous I/O we might need a slightly * different definition. + * + * FIXME: Not sure what different definition we would need? I guess we + * could add the READ_BUFFERS_SYNCHRONOUSLY flag automatically? */ if (max_ios == 0) max_ios = 1; -- 2.48.1.76.g4e746b1a31.dirty --w6dfit2y42fwvotd Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.10-0013-docs-Reframe-track_io_timing-related-docs-as-w.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.11 13/27] aio: Basic read_stream adjustments for real AIO @ 2025-03-18 18:40 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency There is one comment talking about max_ios logic with "real asynchronous I/O" that I am not sure about, so I left it alone for now. There are further improvements we should consider, e.g. waiting to issue IOs until we can issue multiple IOs at once. But that's left for a future change, since it would involve additional heuristics. --- src/backend/storage/aio/read_stream.c | 29 ++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index 18ecf4affc7..60a841816d8 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,7 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + bool sync_mode; bool advice_enabled; bool temporary; @@ -416,6 +418,13 @@ read_stream_start_pending_read(ReadStream *stream) static void read_stream_look_ahead(ReadStream *stream) { + /* + * Allow amortizing the cost of submitting IO over multiple IOs. This + * requires that we don't do any operations that could lead to a deadlock + * with staged-but-unsubmitted IO. + */ + pgaio_enter_batchmode(); + while (stream->ios_in_progress < stream->max_ios && stream->pinned_buffers + stream->pending_read_nblocks < stream->distance) { @@ -463,6 +472,7 @@ read_stream_look_ahead(ReadStream *stream) { /* We've hit the buffer or I/O limit. Rewind and stop here. */ read_stream_unget_block(stream, blocknum); + pgaio_exit_batchmode(); return; } } @@ -497,6 +507,8 @@ read_stream_look_ahead(ReadStream *stream) * time. */ Assert(stream->pinned_buffers > 0 || stream->distance == 0); + + pgaio_exit_batchmode(); } /* @@ -629,15 +641,19 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * This system supports prefetching advice. + * + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; @@ -647,6 +663,9 @@ read_stream_begin_impl(int flags, * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled * above. If we had real asynchronous I/O we might need a slightly * different definition. + * + * FIXME: Not sure what different definition we would need? I guess we + * could add the READ_BUFFERS_SYNCHRONOUSLY flag automatically? */ if (max_ios == 0) max_ios = 1; -- 2.48.1.76.g4e746b1a31.dirty --bjnmbpad43bpmfxt Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.11-0014-docs-Reframe-track_io_timing-related-docs-as-w.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.12 10/28] aio: Basic read_stream adjustments for real AIO @ 2025-03-18 18:40 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO There is one comment talking about max_ios logic with "real asynchronous I/O" that I am not sure about, so I left it alone for now. There are further improvements we should consider: - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency - We could wait to issue IOs until we can issue multiple IOs at once --- src/backend/storage/aio/read_stream.c | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index c60e37e7f7f..df16530d673 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,7 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + bool sync_mode; /* using io_method=sync */ bool advice_enabled; bool temporary; @@ -613,15 +615,19 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * This system supports prefetching advice. + * + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; @@ -631,6 +637,9 @@ read_stream_begin_impl(int flags, * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled * above. If we had real asynchronous I/O we might need a slightly * different definition. + * + * FIXME: Not sure what different definition we would need? I guess we + * could add the READ_BUFFERS_SYNCHRONOUSLY flag automatically? */ if (max_ios == 0) max_ios = 1; -- 2.48.1.76.g4e746b1a31.dirty --5i73spx2p4vwf7fe Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.12-0011-read_stream-Introduce-and-use-optional-batchmo.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.13 13/28] aio: Basic read_stream adjustments for real AIO @ 2025-03-18 18:40 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO - If effective_io_concurrency=0, pass READ_BUFFERS_SYNCHRONOUSLY to StartReadBuffers() to ensure synchronous IO execution There are further improvements we should consider: - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency. That however requires care to avoid deadlocks and thus done separately. - It can be beneficial to defer starting new IOs until we can issue multiple IOs at once. That however requires non-trivial heuristics to decide when to do so. Co-authored-by: Andres Freund <andres@anarazel.de> Co-authored-by: Thomas Munro <thomas.munro@gmail.com> --- src/backend/storage/aio/read_stream.c | 39 ++++++++++++++++++--------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index c60e37e7f7f..26e5dfe77db 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,8 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + int read_buffers_flags; + bool sync_mode; /* using io_method=sync */ bool advice_enabled; bool temporary; @@ -250,7 +253,7 @@ read_stream_start_pending_read(ReadStream *stream) Assert(stream->next_buffer_index == stream->oldest_buffer_index); /* Do we need to issue read-ahead advice? */ - flags = 0; + flags = stream->read_buffers_flags; if (stream->advice_enabled) { if (stream->pending_read_blocknum == stream->seq_blocknum) @@ -261,7 +264,7 @@ read_stream_start_pending_read(ReadStream *stream) * then stay of the way of the kernel's own read-ahead. */ if (stream->seq_until_processed != InvalidBlockNumber) - flags = READ_BUFFERS_ISSUE_ADVICE; + flags |= READ_BUFFERS_ISSUE_ADVICE; } else { @@ -272,7 +275,7 @@ read_stream_start_pending_read(ReadStream *stream) */ stream->seq_until_processed = stream->pending_read_blocknum; if (stream->pinned_buffers > 0) - flags = READ_BUFFERS_ISSUE_ADVICE; + flags |= READ_BUFFERS_ISSUE_ADVICE; } } @@ -613,27 +616,33 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * Read-ahead advice simulating asynchronous I/O with synchronous calls. + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; #endif /* - * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled - * above. If we had real asynchronous I/O we might need a slightly - * different definition. + * Setting max_ios to zero disables AIO and advice-based pseudo AIO, but + * we still need to allocate space to combine and run one I/O. Bump it up + * to one, and remember to ask for synchronous I/O only. */ if (max_ios == 0) + { max_ios = 1; + stream->read_buffers_flags = READ_BUFFERS_SYNCHRONOUSLY; + } /* * Capture stable values for these two GUC-derived numbers for the @@ -777,6 +786,11 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) if (likely(next_blocknum != InvalidBlockNumber)) { + int flags = stream->read_buffers_flags; + + if (stream->advice_enabled) + flags |= READ_BUFFERS_ISSUE_ADVICE; + /* * Pin a buffer for the next call. Same buffer entry, and * arbitrary I/O entry (they're all free). We don't have to @@ -792,8 +806,7 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) if (likely(!StartReadBuffer(&stream->ios[0].op, &stream->buffers[oldest_buffer_index], next_blocknum, - stream->advice_enabled ? - READ_BUFFERS_ISSUE_ADVICE : 0))) + flags))) { /* Fast return. */ return buffer; -- 2.48.1.76.g4e746b1a31.dirty --pro7bqageygxfsvg Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.13-0014-read_stream-Introduce-and-use-optional-batchmo.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* [PATCH v2.14 14/29] aio: Basic read_stream adjustments for real AIO @ 2025-03-18 18:40 Andres Freund <andres@anarazel.de> 0 siblings, 0 replies; 8+ messages in thread From: Andres Freund @ 2025-03-18 18:40 UTC (permalink / raw) Adapt the read stream logic for real AIO: - If AIO is enabled, we shouldn't issue advice, but if it isn't, we should continue issuing advice - AIO benefits from reading ahead with direct IO - If effective_io_concurrency=0, pass READ_BUFFERS_SYNCHRONOUSLY to StartReadBuffers() to ensure synchronous IO execution There are further improvements we should consider: - While in read_stream_look_ahead(), we can use AIO batch submission mode for increased efficiency. That however requires care to avoid deadlocks and thus done separately. - It can be beneficial to defer starting new IOs until we can issue multiple IOs at once. That however requires non-trivial heuristics to decide when to do so. Co-authored-by: Andres Freund <andres@anarazel.de> Co-authored-by: Thomas Munro <thomas.munro@gmail.com> --- src/backend/storage/aio/read_stream.c | 39 ++++++++++++++++++--------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/src/backend/storage/aio/read_stream.c b/src/backend/storage/aio/read_stream.c index c60e37e7f7f..26e5dfe77db 100644 --- a/src/backend/storage/aio/read_stream.c +++ b/src/backend/storage/aio/read_stream.c @@ -72,6 +72,7 @@ #include "postgres.h" #include "miscadmin.h" +#include "storage/aio.h" #include "storage/fd.h" #include "storage/smgr.h" #include "storage/read_stream.h" @@ -99,6 +100,8 @@ struct ReadStream int16 pinned_buffers; int16 distance; int16 initialized_buffers; + int read_buffers_flags; + bool sync_mode; /* using io_method=sync */ bool advice_enabled; bool temporary; @@ -250,7 +253,7 @@ read_stream_start_pending_read(ReadStream *stream) Assert(stream->next_buffer_index == stream->oldest_buffer_index); /* Do we need to issue read-ahead advice? */ - flags = 0; + flags = stream->read_buffers_flags; if (stream->advice_enabled) { if (stream->pending_read_blocknum == stream->seq_blocknum) @@ -261,7 +264,7 @@ read_stream_start_pending_read(ReadStream *stream) * then stay of the way of the kernel's own read-ahead. */ if (stream->seq_until_processed != InvalidBlockNumber) - flags = READ_BUFFERS_ISSUE_ADVICE; + flags |= READ_BUFFERS_ISSUE_ADVICE; } else { @@ -272,7 +275,7 @@ read_stream_start_pending_read(ReadStream *stream) */ stream->seq_until_processed = stream->pending_read_blocknum; if (stream->pinned_buffers > 0) - flags = READ_BUFFERS_ISSUE_ADVICE; + flags |= READ_BUFFERS_ISSUE_ADVICE; } } @@ -613,27 +616,33 @@ read_stream_begin_impl(int flags, stream->per_buffer_data = (void *) MAXALIGN(&stream->ios[Max(1, max_ios)]); + stream->sync_mode = io_method == IOMETHOD_SYNC; + #ifdef USE_PREFETCH /* - * This system supports prefetching advice. We can use it as long as - * direct I/O isn't enabled, the caller hasn't promised sequential access - * (overriding our detection heuristics), and max_ios hasn't been set to - * zero. + * Read-ahead advice simulating asynchronous I/O with synchronous calls. + * Issue advice only if AIO is not used, direct I/O isn't enabled, the + * caller hasn't promised sequential access (overriding our detection + * heuristics), and max_ios hasn't been set to zero. */ - if ((io_direct_flags & IO_DIRECT_DATA) == 0 && + if (stream->sync_mode && + (io_direct_flags & IO_DIRECT_DATA) == 0 && (flags & READ_STREAM_SEQUENTIAL) == 0 && max_ios > 0) stream->advice_enabled = true; #endif /* - * For now, max_ios = 0 is interpreted as max_ios = 1 with advice disabled - * above. If we had real asynchronous I/O we might need a slightly - * different definition. + * Setting max_ios to zero disables AIO and advice-based pseudo AIO, but + * we still need to allocate space to combine and run one I/O. Bump it up + * to one, and remember to ask for synchronous I/O only. */ if (max_ios == 0) + { max_ios = 1; + stream->read_buffers_flags = READ_BUFFERS_SYNCHRONOUSLY; + } /* * Capture stable values for these two GUC-derived numbers for the @@ -777,6 +786,11 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) if (likely(next_blocknum != InvalidBlockNumber)) { + int flags = stream->read_buffers_flags; + + if (stream->advice_enabled) + flags |= READ_BUFFERS_ISSUE_ADVICE; + /* * Pin a buffer for the next call. Same buffer entry, and * arbitrary I/O entry (they're all free). We don't have to @@ -792,8 +806,7 @@ read_stream_next_buffer(ReadStream *stream, void **per_buffer_data) if (likely(!StartReadBuffer(&stream->ios[0].op, &stream->buffers[oldest_buffer_index], next_blocknum, - stream->advice_enabled ? - READ_BUFFERS_ISSUE_ADVICE : 0))) + flags))) { /* Fast return. */ return buffer; -- 2.48.1.76.g4e746b1a31.dirty --m2na7lgr3zfazgom Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v2.14-0015-read_stream-Introduce-and-use-optional-batchmo.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
end of thread, other threads:[~2025-03-18 18:40 UTC | newest] Thread overview: 8+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-12-31 14:19 [PATCH v45 2/3] Filling gaps in jsonb Dmitrii Dolgov <9erthalion6@gmail.com> 2025-03-14 15:39 [PATCH v2.8 26/38] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de> 2025-03-15 16:29 [PATCH v2.9 18/30] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de> 2025-03-18 18:40 [PATCH v2.10 12/28] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de> 2025-03-18 18:40 [PATCH v2.11 13/27] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de> 2025-03-18 18:40 [PATCH v2.12 10/28] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de> 2025-03-18 18:40 [PATCH v2.13 13/28] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de> 2025-03-18 18:40 [PATCH v2.14 14/29] aio: Basic read_stream adjustments for real AIO Andres Freund <andres@anarazel.de>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox