PostgreSQL Source Code  git master
local_source.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * local_source.c
4  * Functions for using a local data directory as the source.
5  *
6  * Portions Copyright (c) 2013-2022, PostgreSQL Global Development Group
7  *
8  *-------------------------------------------------------------------------
9  */
10 #include "postgres_fe.h"
11 
12 #include <fcntl.h>
13 #include <unistd.h>
14 
15 #include "datapagemap.h"
16 #include "file_ops.h"
17 #include "filemap.h"
18 #include "pg_rewind.h"
19 #include "rewind_source.h"
20 
21 typedef struct
22 {
23  rewind_source common; /* common interface functions */
24 
25  const char *datadir; /* path to the source data directory */
26 } local_source;
27 
30 static char *local_fetch_file(rewind_source *source, const char *path,
31  size_t *filesize);
32 static void local_queue_fetch_file(rewind_source *source, const char *path,
33  size_t len);
34 static void local_queue_fetch_range(rewind_source *source, const char *path,
35  off_t off, size_t len);
37 static void local_destroy(rewind_source *source);
38 
41 {
42  local_source *src;
43 
44  src = pg_malloc0(sizeof(local_source));
45 
53 
54  src->datadir = datadir;
55 
56  return &src->common;
57 }
58 
59 static void
61 {
63 }
64 
65 static char *
66 local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
67 {
68  return slurpFile(((local_source *) source)->datadir, path, filesize);
69 }
70 
71 /*
72  * Copy a file from source to target.
73  *
74  * 'len' is the expected length of the file.
75  */
76 static void
77 local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
78 {
79  const char *datadir = ((local_source *) source)->datadir;
81  char srcpath[MAXPGPATH];
82  int srcfd;
83  size_t written_len;
84 
85  snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
86 
87  /* Open source file for reading */
88  srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
89  if (srcfd < 0)
90  pg_fatal("could not open source file \"%s\": %m",
91  srcpath);
92 
93  /* Truncate and open the target file for writing */
94  open_target_file(path, true);
95 
96  written_len = 0;
97  for (;;)
98  {
99  ssize_t read_len;
100 
101  read_len = read(srcfd, buf.data, sizeof(buf));
102 
103  if (read_len < 0)
104  pg_fatal("could not read file \"%s\": %m", srcpath);
105  else if (read_len == 0)
106  break; /* EOF reached */
107 
108  write_target_range(buf.data, written_len, read_len);
109  written_len += read_len;
110  }
111 
112  /*
113  * A local source is not expected to change while we're rewinding, so
114  * check that the size of the file matches our earlier expectation.
115  */
116  if (written_len != len)
117  pg_fatal("size of source file \"%s\" changed concurrently: %d bytes expected, %d copied",
118  srcpath, (int) len, (int) written_len);
119 
120  if (close(srcfd) != 0)
121  pg_fatal("could not close file \"%s\": %m", srcpath);
122 }
123 
124 /*
125  * Copy a file from source to target, starting at 'off', for 'len' bytes.
126  */
127 static void
128 local_queue_fetch_range(rewind_source *source, const char *path, off_t off,
129  size_t len)
130 {
131  const char *datadir = ((local_source *) source)->datadir;
133  char srcpath[MAXPGPATH];
134  int srcfd;
135  off_t begin = off;
136  off_t end = off + len;
137 
138  snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
139 
140  srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
141  if (srcfd < 0)
142  pg_fatal("could not open source file \"%s\": %m",
143  srcpath);
144 
145  if (lseek(srcfd, begin, SEEK_SET) == -1)
146  pg_fatal("could not seek in source file: %m");
147 
148  open_target_file(path, false);
149 
150  while (end - begin > 0)
151  {
152  ssize_t readlen;
153  size_t thislen;
154 
155  if (end - begin > sizeof(buf))
156  thislen = sizeof(buf);
157  else
158  thislen = end - begin;
159 
160  readlen = read(srcfd, buf.data, thislen);
161 
162  if (readlen < 0)
163  pg_fatal("could not read file \"%s\": %m", srcpath);
164  else if (readlen == 0)
165  pg_fatal("unexpected EOF while reading file \"%s\"", srcpath);
166 
167  write_target_range(buf.data, begin, readlen);
168  begin += readlen;
169  }
170 
171  if (close(srcfd) != 0)
172  pg_fatal("could not close file \"%s\": %m", srcpath);
173 }
174 
175 static void
177 {
178  /*
179  * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
180  */
181 }
182 
183 static void
185 {
186  pfree(source);
187 }
#define PG_BINARY
Definition: c.h:1279
void * pg_malloc0(size_t size)
Definition: fe_memutils.c:53
void traverse_datadir(const char *datadir, process_file_callback_t callback)
Definition: file_ops.c:362
char * slurpFile(const char *datadir, const char *path, size_t *filesize)
Definition: file_ops.c:314
void open_target_file(const char *path, bool trunc)
Definition: file_ops.c:47
void write_target_range(char *buf, off_t begin, size_t size)
Definition: file_ops.c:88
void(* process_file_callback_t)(const char *path, file_type_t type, size_t size, const char *link_target)
Definition: file_ops.h:26
void process_source_file(const char *path, file_type_t type, size_t size, const char *link_target)
Definition: filemap.c:218
#define close(a)
Definition: win32.h:12
#define read(a, b, c)
Definition: win32.h:13
static void local_queue_fetch_file(rewind_source *source, const char *path, size_t len)
Definition: local_source.c:77
static void local_destroy(rewind_source *source)
Definition: local_source.c:184
rewind_source * init_local_source(const char *datadir)
Definition: local_source.c:40
static void local_traverse_files(rewind_source *source, process_file_callback_t callback)
Definition: local_source.c:60
static char * local_fetch_file(rewind_source *source, const char *path, size_t *filesize)
Definition: local_source.c:66
static void local_finish_fetch(rewind_source *source)
Definition: local_source.c:176
static void local_queue_fetch_range(rewind_source *source, const char *path, off_t off, size_t len)
Definition: local_source.c:128
void pfree(void *pointer)
Definition: mcxt.c:1175
#define pg_fatal(...)
#define MAXPGPATH
const void size_t len
char * datadir
static rewind_source * source
Definition: pg_rewind.c:81
static char * buf
Definition: pg_test_fsync.c:67
#define snprintf
Definition: port.h:225
rewind_source common
Definition: local_source.c:23
const char * datadir
Definition: local_source.c:25
void(* queue_fetch_file)(struct rewind_source *, const char *path, size_t len)
Definition: rewind_source.h:60
void(* traverse_files)(struct rewind_source *, process_file_callback_t callback)
Definition: rewind_source.h:29
void(* finish_fetch)(struct rewind_source *)
Definition: rewind_source.h:66
XLogRecPtr(* get_current_wal_insert_lsn)(struct rewind_source *)
Definition: rewind_source.h:71
void(* queue_fetch_range)(struct rewind_source *, const char *path, off_t offset, size_t len)
Definition: rewind_source.h:47
char *(* fetch_file)(struct rewind_source *, const char *path, size_t *filesize)
Definition: rewind_source.h:37
void(* destroy)(struct rewind_source *)
Definition: rewind_source.h:76
static void callback(struct sockaddr *addr, struct sockaddr *mask, void *unused)
Definition: test_ifaddrs.c:48