• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 // Protocol Buffers - Google's data interchange format
2 // Copyright 2008 Google Inc.  All rights reserved.
3 // https://developers.google.com/protocol-buffers/
4 //
5 // Redistribution and use in source and binary forms, with or without
6 // modification, are permitted provided that the following conditions are
7 // met:
8 //
9 //     * Redistributions of source code must retain the above copyright
10 // notice, this list of conditions and the following disclaimer.
11 //     * Redistributions in binary form must reproduce the above
12 // copyright notice, this list of conditions and the following disclaimer
13 // in the documentation and/or other materials provided with the
14 // distribution.
15 //     * Neither the name of Google Inc. nor the names of its
16 // contributors may be used to endorse or promote products derived from
17 // this software without specific prior written permission.
18 //
19 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
20 // "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
21 // LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
22 // A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
23 // OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
24 // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
25 // LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
26 // DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
27 // THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
28 // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
29 // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
30 
31 // Author: kenton@google.com (Kenton Varda)
32 //  Based on original Protocol Buffers design by
33 //  Sanjay Ghemawat, Jeff Dean, and others.
34 
35 #ifndef _MSC_VER
36 #include <fcntl.h>
37 #include <sys/stat.h>
38 #include <sys/types.h>
39 #include <unistd.h>
40 #endif
41 #include <errno.h>
42 
43 #include <algorithm>
44 #include <iostream>
45 
46 #include <google/protobuf/stubs/common.h>
47 #include <google/protobuf/stubs/logging.h>
48 #include <google/protobuf/io/io_win32.h>
49 #include <google/protobuf/io/zero_copy_stream_impl.h>
50 #include <google/protobuf/stubs/stl_util.h>
51 
52 
53 namespace google {
54 namespace protobuf {
55 namespace io {
56 
57 #ifdef _WIN32
58 // Win32 lseek is broken:  If invoked on a non-seekable file descriptor, its
59 // return value is undefined.  We re-define it to always produce an error.
60 #define lseek(fd, offset, origin) ((off_t)-1)
61 // DO NOT include <io.h>, instead create functions in io_win32.{h,cc} and import
62 // them like we do below.
63 using google::protobuf::io::win32::access;
64 using google::protobuf::io::win32::close;
65 using google::protobuf::io::win32::open;
66 using google::protobuf::io::win32::read;
67 using google::protobuf::io::win32::write;
68 #endif
69 
70 namespace {
71 
72 // EINTR sucks.
close_no_eintr(int fd)73 int close_no_eintr(int fd) {
74   int result;
75   do {
76     result = close(fd);
77   } while (result < 0 && errno == EINTR);
78   return result;
79 }
80 
81 }  // namespace
82 
83 // ===================================================================
84 
FileInputStream(int file_descriptor,int block_size)85 FileInputStream::FileInputStream(int file_descriptor, int block_size)
86     : copying_input_(file_descriptor), impl_(&copying_input_, block_size) {}
87 
Close()88 bool FileInputStream::Close() { return copying_input_.Close(); }
89 
Next(const void ** data,int * size)90 bool FileInputStream::Next(const void** data, int* size) {
91   return impl_.Next(data, size);
92 }
93 
BackUp(int count)94 void FileInputStream::BackUp(int count) { impl_.BackUp(count); }
95 
Skip(int count)96 bool FileInputStream::Skip(int count) { return impl_.Skip(count); }
97 
ByteCount() const98 int64_t FileInputStream::ByteCount() const { return impl_.ByteCount(); }
99 
CopyingFileInputStream(int file_descriptor)100 FileInputStream::CopyingFileInputStream::CopyingFileInputStream(
101     int file_descriptor)
102     : file_(file_descriptor),
103       close_on_delete_(false),
104       is_closed_(false),
105       errno_(0),
106       previous_seek_failed_(false) {}
107 
~CopyingFileInputStream()108 FileInputStream::CopyingFileInputStream::~CopyingFileInputStream() {
109   if (close_on_delete_) {
110     if (!Close()) {
111       GOOGLE_LOG(ERROR) << "close() failed: " << strerror(errno_);
112     }
113   }
114 }
115 
Close()116 bool FileInputStream::CopyingFileInputStream::Close() {
117   GOOGLE_CHECK(!is_closed_);
118 
119   is_closed_ = true;
120   if (close_no_eintr(file_) != 0) {
121     // The docs on close() do not specify whether a file descriptor is still
122     // open after close() fails with EIO.  However, the glibc source code
123     // seems to indicate that it is not.
124     errno_ = errno;
125     return false;
126   }
127 
128   return true;
129 }
130 
Read(void * buffer,int size)131 int FileInputStream::CopyingFileInputStream::Read(void* buffer, int size) {
132   GOOGLE_CHECK(!is_closed_);
133 
134   int result;
135   do {
136     result = read(file_, buffer, size);
137   } while (result < 0 && errno == EINTR);
138 
139   if (result < 0) {
140     // Read error (not EOF).
141     errno_ = errno;
142   }
143 
144   return result;
145 }
146 
Skip(int count)147 int FileInputStream::CopyingFileInputStream::Skip(int count) {
148   GOOGLE_CHECK(!is_closed_);
149 
150   if (!previous_seek_failed_ && lseek(file_, count, SEEK_CUR) != (off_t)-1) {
151     // Seek succeeded.
152     return count;
153   } else {
154     // Failed to seek.
155 
156     // Note to self:  Don't seek again.  This file descriptor doesn't
157     // support it.
158     previous_seek_failed_ = true;
159 
160     // Use the default implementation.
161     return CopyingInputStream::Skip(count);
162   }
163 }
164 
165 // ===================================================================
166 
FileOutputStream(int file_descriptor,int block_size)167 FileOutputStream::FileOutputStream(int file_descriptor, int block_size)
168     : CopyingOutputStreamAdaptor(&copying_output_),
169       copying_output_(file_descriptor) {}
170 
Close()171 bool FileOutputStream::Close() {
172   bool flush_succeeded = Flush();
173   return copying_output_.Close() && flush_succeeded;
174 }
175 
CopyingFileOutputStream(int file_descriptor)176 FileOutputStream::CopyingFileOutputStream::CopyingFileOutputStream(
177     int file_descriptor)
178     : file_(file_descriptor),
179       close_on_delete_(false),
180       is_closed_(false),
181       errno_(0) {}
182 
~FileOutputStream()183 FileOutputStream::~FileOutputStream() { Flush(); }
184 
~CopyingFileOutputStream()185 FileOutputStream::CopyingFileOutputStream::~CopyingFileOutputStream() {
186   if (close_on_delete_) {
187     if (!Close()) {
188       GOOGLE_LOG(ERROR) << "close() failed: " << strerror(errno_);
189     }
190   }
191 }
192 
Close()193 bool FileOutputStream::CopyingFileOutputStream::Close() {
194   GOOGLE_CHECK(!is_closed_);
195 
196   is_closed_ = true;
197   if (close_no_eintr(file_) != 0) {
198     // The docs on close() do not specify whether a file descriptor is still
199     // open after close() fails with EIO.  However, the glibc source code
200     // seems to indicate that it is not.
201     errno_ = errno;
202     return false;
203   }
204 
205   return true;
206 }
207 
Write(const void * buffer,int size)208 bool FileOutputStream::CopyingFileOutputStream::Write(const void* buffer,
209                                                       int size) {
210   GOOGLE_CHECK(!is_closed_);
211   int total_written = 0;
212 
213   const uint8* buffer_base = reinterpret_cast<const uint8*>(buffer);
214 
215   while (total_written < size) {
216     int bytes;
217     do {
218       bytes = write(file_, buffer_base + total_written, size - total_written);
219     } while (bytes < 0 && errno == EINTR);
220 
221     if (bytes <= 0) {
222       // Write error.
223 
224       // FIXME(kenton):  According to the man page, if write() returns zero,
225       //   there was no error; write() simply did not write anything.  It's
226       //   unclear under what circumstances this might happen, but presumably
227       //   errno won't be set in this case.  I am confused as to how such an
228       //   event should be handled.  For now I'm treating it as an error, since
229       //   retrying seems like it could lead to an infinite loop.  I suspect
230       //   this never actually happens anyway.
231 
232       if (bytes < 0) {
233         errno_ = errno;
234       }
235       return false;
236     }
237     total_written += bytes;
238   }
239 
240   return true;
241 }
242 
243 // ===================================================================
244 
IstreamInputStream(std::istream * input,int block_size)245 IstreamInputStream::IstreamInputStream(std::istream* input, int block_size)
246     : copying_input_(input), impl_(&copying_input_, block_size) {}
247 
Next(const void ** data,int * size)248 bool IstreamInputStream::Next(const void** data, int* size) {
249   return impl_.Next(data, size);
250 }
251 
BackUp(int count)252 void IstreamInputStream::BackUp(int count) { impl_.BackUp(count); }
253 
Skip(int count)254 bool IstreamInputStream::Skip(int count) { return impl_.Skip(count); }
255 
ByteCount() const256 int64_t IstreamInputStream::ByteCount() const { return impl_.ByteCount(); }
257 
CopyingIstreamInputStream(std::istream * input)258 IstreamInputStream::CopyingIstreamInputStream::CopyingIstreamInputStream(
259     std::istream* input)
260     : input_(input) {}
261 
~CopyingIstreamInputStream()262 IstreamInputStream::CopyingIstreamInputStream::~CopyingIstreamInputStream() {}
263 
Read(void * buffer,int size)264 int IstreamInputStream::CopyingIstreamInputStream::Read(void* buffer,
265                                                         int size) {
266   input_->read(reinterpret_cast<char*>(buffer), size);
267   int result = input_->gcount();
268   if (result == 0 && input_->fail() && !input_->eof()) {
269     return -1;
270   }
271   return result;
272 }
273 
274 // ===================================================================
275 
OstreamOutputStream(std::ostream * output,int block_size)276 OstreamOutputStream::OstreamOutputStream(std::ostream* output, int block_size)
277     : copying_output_(output), impl_(&copying_output_, block_size) {}
278 
~OstreamOutputStream()279 OstreamOutputStream::~OstreamOutputStream() { impl_.Flush(); }
280 
Next(void ** data,int * size)281 bool OstreamOutputStream::Next(void** data, int* size) {
282   return impl_.Next(data, size);
283 }
284 
BackUp(int count)285 void OstreamOutputStream::BackUp(int count) { impl_.BackUp(count); }
286 
ByteCount() const287 int64_t OstreamOutputStream::ByteCount() const { return impl_.ByteCount(); }
288 
CopyingOstreamOutputStream(std::ostream * output)289 OstreamOutputStream::CopyingOstreamOutputStream::CopyingOstreamOutputStream(
290     std::ostream* output)
291     : output_(output) {}
292 
~CopyingOstreamOutputStream()293 OstreamOutputStream::CopyingOstreamOutputStream::~CopyingOstreamOutputStream() {
294 }
295 
Write(const void * buffer,int size)296 bool OstreamOutputStream::CopyingOstreamOutputStream::Write(const void* buffer,
297                                                             int size) {
298   output_->write(reinterpret_cast<const char*>(buffer), size);
299   return output_->good();
300 }
301 
302 // ===================================================================
303 
ConcatenatingInputStream(ZeroCopyInputStream * const streams[],int count)304 ConcatenatingInputStream::ConcatenatingInputStream(
305     ZeroCopyInputStream* const streams[], int count)
306     : streams_(streams), stream_count_(count), bytes_retired_(0) {
307 }
308 
Next(const void ** data,int * size)309 bool ConcatenatingInputStream::Next(const void** data, int* size) {
310   while (stream_count_ > 0) {
311     if (streams_[0]->Next(data, size)) return true;
312 
313     // That stream is done.  Advance to the next one.
314     bytes_retired_ += streams_[0]->ByteCount();
315     ++streams_;
316     --stream_count_;
317   }
318 
319   // No more streams.
320   return false;
321 }
322 
BackUp(int count)323 void ConcatenatingInputStream::BackUp(int count) {
324   if (stream_count_ > 0) {
325     streams_[0]->BackUp(count);
326   } else {
327     GOOGLE_LOG(DFATAL) << "Can't BackUp() after failed Next().";
328   }
329 }
330 
Skip(int count)331 bool ConcatenatingInputStream::Skip(int count) {
332   while (stream_count_ > 0) {
333     // Assume that ByteCount() can be used to find out how much we actually
334     // skipped when Skip() fails.
335     int64 target_byte_count = streams_[0]->ByteCount() + count;
336     if (streams_[0]->Skip(count)) return true;
337 
338     // Hit the end of the stream.  Figure out how many more bytes we still have
339     // to skip.
340     int64 final_byte_count = streams_[0]->ByteCount();
341     GOOGLE_DCHECK_LT(final_byte_count, target_byte_count);
342     count = target_byte_count - final_byte_count;
343 
344     // That stream is done.  Advance to the next one.
345     bytes_retired_ += final_byte_count;
346     ++streams_;
347     --stream_count_;
348   }
349 
350   return false;
351 }
352 
ByteCount() const353 int64_t ConcatenatingInputStream::ByteCount() const {
354   if (stream_count_ == 0) {
355     return bytes_retired_;
356   } else {
357     return bytes_retired_ + streams_[0]->ByteCount();
358   }
359 }
360 
361 
362 // ===================================================================
363 
364 }  // namespace io
365 }  // namespace protobuf
366 }  // namespace google
367