Re: Parallel copy

Hi,

On 2020-06-03 15:53:24 +0530, vignesh C wrote:
> Workers/
> Exec time (seconds) copy from file,
> 2 indexes on integer columns
> 1 index on text column copy from stdin,
> 2 indexes on integer columns
> 1 index on text column copy from file, 1 gist index on text column copy
> from file,
> 3 indexes on integer columns copy from stdin, 3 indexes on integer columns
> 0 1162.772(1X) 1176.035(1X) 827.669(1X) 216.171(1X) 217.376(1X)
> 1 1110.288(1.05X) 1120.556(1.05X) 747.384(1.11X) 174.242(1.24X) 163.492(1.33X)
> 2 635.249(1.83X) 668.18(1.76X) 435.673(1.9X) 133.829(1.61X) 126.516(1.72X)
> 4 336.835(3.45X) 346.768(3.39X) 236.406(3.5X) 105.767(2.04X) 107.382(2.02X)
> 8 188.577(6.17X) 194.491(6.04X) 148.962(5.56X) 100.708(2.15X) 107.72(2.01X)
> 16 126.819(9.17X) 146.402(8.03X) 119.923(6.9X) 97.996(2.2X) 106.531(2.04X)
> 20 *117.845(9.87X)* 149.203(7.88X) 138.741(5.96X) 97.94(2.21X) 107.5(2.02)
> 30 127.554(9.11X) 161.218(7.29X) 172.443(4.8X) 98.232(2.2X) 108.778(1.99X)

Hm. you don't explicitly mention that in your design, but given how
small the benefits going from 0-1 workers is, I assume the leader
doesn't do any "chunk processing" on its own?

> Design of the Parallel Copy: The backend, to which the "COPY FROM" query is
> submitted acts as leader with the responsibility of reading data from the
> file/stdin, launching at most n number of workers as specified with
> PARALLEL 'n' option in the "COPY FROM" query. The leader populates the
> common data required for the workers execution in the DSM and shares it
> with the workers. The leader then executes before statement triggers if
> there exists any. Leader populates DSM chunks which includes the start
> offset and chunk size, while populating the chunks it reads as many blocks
> as required into the DSM data blocks from the file. Each block is of 64K
> size. The leader parses the data to identify a chunk, the existing logic
> from CopyReadLineText which identifies the chunks with some changes was
> used for this. Leader checks if a free chunk is available to copy the
> information, if there is no free chunk it waits till the required chunk is
> freed up by the worker and then copies the identified chunks information
> (offset & chunk size) into the DSM chunks. This process is repeated till
> the complete file is processed. Simultaneously, the workers cache the
> chunks(50) locally into the local memory and release the chunks to the
> leader for further populating. Each worker processes the chunk that it
> cached and inserts it into the table. The leader waits till all the chunks
> populated are processed by the workers and exits.

Why do we need the local copy of 50 chunks? Copying memory around is far
from free. I don't see why it'd be better to add per-process caching,
rather than making the DSM bigger? I can see some benefit in marking
multiple chunks as being processed with one lock acquisition, but I
don't think adding a memory copy is a good idea.

This patch *desperately* needs to be split up. It imo is close to
unreviewable, due to a large amount of changes that just move code
around without other functional changes being mixed in with the actual
new stuff.

> /*
> + * State of the chunk.
> + */
> +typedef enum ChunkState
> +{
> + CHUNK_INIT, /* initial state of chunk */
> + CHUNK_LEADER_POPULATING, /* leader processing chunk */
> + CHUNK_LEADER_POPULATED, /* leader completed populating chunk */
> + CHUNK_WORKER_PROCESSING, /* worker processing chunk */
> + CHUNK_WORKER_PROCESSED /* worker completed processing chunk */
> +}ChunkState;
> +
> +#define RAW_BUF_SIZE 65536 /* we palloc RAW_BUF_SIZE+1 bytes */
> +
> +#define DATA_BLOCK_SIZE RAW_BUF_SIZE
> +#define RINGSIZE (10 * 1000)
> +#define MAX_BLOCKS_COUNT 1000
> +#define WORKER_CHUNK_COUNT 50 /* should be mod of RINGSIZE */
> +
> +#define IsParallelCopy() (cstate->is_parallel)
> +#define IsLeader() (cstate->pcdata->is_leader)
> +#define IsHeaderLine() (cstate->header_line && cstate->cur_lineno == 1)
> +
> +/*
> + * Copy data block information.
> + */
> +typedef struct CopyDataBlock
> +{
> + /* The number of unprocessed chunks in the current block. */
> + pg_atomic_uint32 unprocessed_chunk_parts;
> +
> + /*
> + * If the current chunk data is continued into another block,
> + * following_block will have the position where the remaining data need to
> + * be read.
> + */
> + uint32 following_block;
> +
> + /*
> + * This flag will be set, when the leader finds out this block can be read
> + * safely by the worker. This helps the worker to start processing the chunk
> + * early where the chunk will be spread across many blocks and the worker
> + * need not wait for the complete chunk to be processed.
> + */
> + bool curr_blk_completed;
> + char data[DATA_BLOCK_SIZE + 1]; /* data read from file */
> +}CopyDataBlock;

What's the + 1 here about?

> +/*
> + * Parallel copy line buffer information.
> + */
> +typedef struct ParallelCopyLineBuf
> +{
> + StringInfoData line_buf;
> + uint64 cur_lineno; /* line number for error messages */
> +}ParallelCopyLineBuf;

Why do we need separate infrastructure for this? We shouldn't duplicate
infrastructure unnecessarily.

> +/*
> + * Common information that need to be copied to shared memory.
> + */
> +typedef struct CopyWorkerCommonData
> +{

Why is parallel specific stuff here suddenly not named ParallelCopy*
anymore? If you introduce a naming like that it imo should be used
consistently.

> + /* low-level state data */
> + CopyDest copy_dest; /* type of copy source/destination */
> + int file_encoding; /* file or remote side's character encoding */
> + bool need_transcoding; /* file encoding diff from server? */
> + bool encoding_embeds_ascii; /* ASCII can be non-first byte? */
> +
> + /* parameters from the COPY command */
> + bool csv_mode; /* Comma Separated Value format? */
> + bool header_line; /* CSV header line? */
> + int null_print_len; /* length of same */
> + bool force_quote_all; /* FORCE_QUOTE *? */
> + bool convert_selectively; /* do selective binary conversion? */
> +
> + /* Working state for COPY FROM */
> + AttrNumber num_defaults;
> + Oid relid;
> +}CopyWorkerCommonData;

But I actually think we shouldn't have this information in two different
structs. This should exist once, independent of using parallel /
non-parallel copy.

> +/* List information */
> +typedef struct ListInfo
> +{
> + int count; /* count of attributes */
> +
> + /* string info in the form info followed by info1, info2... infon */
> + char info[1];
> +} ListInfo;

Based on these comments I have no idea what this could be for.

> /*
> - * This keeps the character read at the top of the loop in the buffer
> - * even if there is more than one read-ahead.
> + * This keeps the character read at the top of the loop in the buffer
> + * even if there is more than one read-ahead.
> + */
> +#define IF_NEED_REFILL_AND_NOT_EOF_CONTINUE(extralen) \
> +if (1) \
> +{ \
> + if (copy_buff_state.raw_buf_ptr + (extralen) >= copy_buff_state.copy_buf_len && !hit_eof) \
> + { \
> + if (IsParallelCopy()) \
> + { \
> + copy_buff_state.chunk_size = prev_chunk_size; /* update previous chunk size */ \
> + if (copy_buff_state.block_switched) \
> + { \
> + pg_atomic_sub_fetch_u32(&copy_buff_state.data_blk_ptr->unprocessed_chunk_parts, 1); \
> + copy_buff_state.copy_buf_len = prev_copy_buf_len; \
> + } \
> + } \
> + copy_buff_state.raw_buf_ptr = prev_raw_ptr; /* undo fetch */ \
> + need_data = true; \
> + continue; \
> + } \
> +} else ((void) 0)

I think it's an absolutely clear no-go to add new branches to
these. They're *really* hot already, and this is going to sprinkle a
significant amount of new instructions over a lot of places.

> +/*
> + * SET_RAWBUF_FOR_LOAD - Set raw_buf to the shared memory where the file data must
> + * be read.
> + */
> +#define SET_RAWBUF_FOR_LOAD() \
> +{ \
> + ShmCopyInfo *pcshared_info = cstate->pcdata->pcshared_info; \
> + uint32 cur_block_pos; \
> + /* \
> + * Mark the previous block as completed, worker can start copying this data. \
> + */ \
> + if (copy_buff_state.data_blk_ptr != copy_buff_state.curr_data_blk_ptr && \
> + copy_buff_state.data_blk_ptr->curr_blk_completed == false) \
> + copy_buff_state.data_blk_ptr->curr_blk_completed = true; \
> + \
> + copy_buff_state.data_blk_ptr = copy_buff_state.curr_data_blk_ptr; \
> + cur_block_pos = WaitGetFreeCopyBlock(pcshared_info); \
> + copy_buff_state.curr_data_blk_ptr = &pcshared_info->data_blocks[cur_block_pos]; \
> + \
> + if (!copy_buff_state.data_blk_ptr) \
> + { \
> + copy_buff_state.data_blk_ptr = copy_buff_state.curr_data_blk_ptr; \
> + chunk_first_block = cur_block_pos; \
> + } \
> + else if (need_data == false) \
> + copy_buff_state.data_blk_ptr->following_block = cur_block_pos; \
> + \
> + cstate->raw_buf = copy_buff_state.curr_data_blk_ptr->data; \
> + copy_buff_state.copy_raw_buf = cstate->raw_buf; \
> +}
> +
> +/*
> + * END_CHUNK_PARALLEL_COPY - Update the chunk information in shared memory.
> + */
> +#define END_CHUNK_PARALLEL_COPY() \
> +{ \
> + if (!IsHeaderLine()) \
> + { \
> + ShmCopyInfo *pcshared_info = cstate->pcdata->pcshared_info; \
> + ChunkBoundaries *chunkBoundaryPtr = &pcshared_info->chunk_boundaries; \
> + if (copy_buff_state.chunk_size) \
> + { \
> + ChunkBoundary *chunkInfo = &chunkBoundaryPtr->ring[chunk_pos]; \
> + /* \
> + * If raw_buf_ptr is zero, unprocessed_chunk_parts would have been \
> + * incremented in SEEK_COPY_BUFF_POS. This will happen if the whole \
> + * chunk finishes at the end of the current block. If the \
> + * new_line_size > raw_buf_ptr, then the new block has only new line \
> + * char content. The unprocessed count should not be increased in \
> + * this case. \
> + */ \
> + if (copy_buff_state.raw_buf_ptr != 0 && \
> + copy_buff_state.raw_buf_ptr > new_line_size) \
> + pg_atomic_add_fetch_u32(&copy_buff_state.curr_data_blk_ptr->unprocessed_chunk_parts, 1); \
> + \
> + /* Update chunk size. */ \
> + pg_atomic_write_u32(&chunkInfo->chunk_size, copy_buff_state.chunk_size); \
> + pg_atomic_write_u32(&chunkInfo->chunk_state, CHUNK_LEADER_POPULATED); \
> + elog(DEBUG1, "[Leader] After adding - chunk position:%d, chunk_size:%d", \
> + chunk_pos, copy_buff_state.chunk_size); \
> + pcshared_info->populated++; \
> + } \
> + else if (new_line_size) \
> + { \
> + /* \
> + * This means only new line char, empty record should be \
> + * inserted. \
> + */ \
> + ChunkBoundary *chunkInfo; \
> + chunk_pos = UpdateBlockInChunkInfo(cstate, -1, -1, 0, \
> + CHUNK_LEADER_POPULATED); \
> + chunkInfo = &chunkBoundaryPtr->ring[chunk_pos]; \
> + elog(DEBUG1, "[Leader] Added empty chunk with offset:%d, chunk position:%d, chunk size:%d", \
> + chunkInfo->start_offset, chunk_pos, \
> + pg_atomic_read_u32(&chunkInfo->chunk_size)); \
> + pcshared_info->populated++; \
> + } \
> + }\
> + \
> + /*\
> + * All of the read data is processed, reset index & len. In the\
> + * subsequent read, we will get a new block and copy data in to the\
> + * new block.\
> + */\
> + if (copy_buff_state.raw_buf_ptr == copy_buff_state.copy_buf_len)\
> + {\
> + cstate->raw_buf_index = 0;\
> + cstate->raw_buf_len = 0;\
> + }\
> + else\
> + cstate->raw_buf_len = copy_buff_state.copy_buf_len;\
> +}

Why are these macros? They are way way way above a length where that
makes any sort of sense.

Greetings,

Andres Freund