• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 /*
2  *
3  * Copyright 2015 gRPC authors.
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  *
17  */
18 
19 #include <grpc/support/port_platform.h>
20 
21 #include "src/core/lib/transport/byte_stream.h"
22 
23 #include <stdlib.h>
24 #include <string.h>
25 
26 #include <grpc/support/log.h>
27 
28 #include "src/core/lib/gprpp/memory.h"
29 #include "src/core/lib/slice/slice_internal.h"
30 
31 namespace grpc_core {
32 
33 //
34 // SliceBufferByteStream
35 //
36 
SliceBufferByteStream(grpc_slice_buffer * slice_buffer,uint32_t flags)37 SliceBufferByteStream::SliceBufferByteStream(grpc_slice_buffer* slice_buffer,
38                                              uint32_t flags)
39     : ByteStream(static_cast<uint32_t>(slice_buffer->length), flags) {
40   GPR_ASSERT(slice_buffer->length <= UINT32_MAX);
41   grpc_slice_buffer_init(&backing_buffer_);
42   grpc_slice_buffer_swap(slice_buffer, &backing_buffer_);
43 }
44 
~SliceBufferByteStream()45 SliceBufferByteStream::~SliceBufferByteStream() {}
46 
Orphan()47 void SliceBufferByteStream::Orphan() {
48   grpc_slice_buffer_destroy_internal(&backing_buffer_);
49   GRPC_ERROR_UNREF(shutdown_error_);
50   // Note: We do not actually delete the object here, since
51   // SliceBufferByteStream is usually allocated as part of a larger
52   // object and has an OrphanablePtr of itself passed down through the
53   // filter stack.
54 }
55 
Next(size_t max_size_hint,grpc_closure * on_complete)56 bool SliceBufferByteStream::Next(size_t max_size_hint,
57                                  grpc_closure* on_complete) {
58   GPR_ASSERT(cursor_ < backing_buffer_.count);
59   return true;
60 }
61 
Pull(grpc_slice * slice)62 grpc_error* SliceBufferByteStream::Pull(grpc_slice* slice) {
63   if (shutdown_error_ != GRPC_ERROR_NONE) {
64     return GRPC_ERROR_REF(shutdown_error_);
65   }
66   GPR_ASSERT(cursor_ < backing_buffer_.count);
67   *slice = grpc_slice_ref_internal(backing_buffer_.slices[cursor_]);
68   ++cursor_;
69   return GRPC_ERROR_NONE;
70 }
71 
Shutdown(grpc_error * error)72 void SliceBufferByteStream::Shutdown(grpc_error* error) {
73   GRPC_ERROR_UNREF(shutdown_error_);
74   shutdown_error_ = error;
75 }
76 
77 //
78 // ByteStreamCache
79 //
80 
ByteStreamCache(OrphanablePtr<ByteStream> underlying_stream)81 ByteStreamCache::ByteStreamCache(OrphanablePtr<ByteStream> underlying_stream)
82     : underlying_stream_(std::move(underlying_stream)),
83       length_(underlying_stream_->length()),
84       flags_(underlying_stream_->flags()) {
85   grpc_slice_buffer_init(&cache_buffer_);
86 }
87 
~ByteStreamCache()88 ByteStreamCache::~ByteStreamCache() { Destroy(); }
89 
Destroy()90 void ByteStreamCache::Destroy() {
91   underlying_stream_.reset();
92   if (cache_buffer_.length > 0) {
93     grpc_slice_buffer_destroy_internal(&cache_buffer_);
94   }
95 }
96 
97 //
98 // ByteStreamCache::CachingByteStream
99 //
100 
CachingByteStream(ByteStreamCache * cache)101 ByteStreamCache::CachingByteStream::CachingByteStream(ByteStreamCache* cache)
102     : ByteStream(cache->length_, cache->flags_), cache_(cache) {}
103 
~CachingByteStream()104 ByteStreamCache::CachingByteStream::~CachingByteStream() {}
105 
Orphan()106 void ByteStreamCache::CachingByteStream::Orphan() {
107   GRPC_ERROR_UNREF(shutdown_error_);
108   // Note: We do not actually delete the object here, since
109   // CachingByteStream is usually allocated as part of a larger
110   // object and has an OrphanablePtr of itself passed down through the
111   // filter stack.
112 }
113 
Next(size_t max_size_hint,grpc_closure * on_complete)114 bool ByteStreamCache::CachingByteStream::Next(size_t max_size_hint,
115                                               grpc_closure* on_complete) {
116   if (shutdown_error_ != GRPC_ERROR_NONE) return true;
117   if (cursor_ < cache_->cache_buffer_.count) return true;
118   GPR_ASSERT(cache_->underlying_stream_ != nullptr);
119   return cache_->underlying_stream_->Next(max_size_hint, on_complete);
120 }
121 
Pull(grpc_slice * slice)122 grpc_error* ByteStreamCache::CachingByteStream::Pull(grpc_slice* slice) {
123   if (shutdown_error_ != GRPC_ERROR_NONE) {
124     return GRPC_ERROR_REF(shutdown_error_);
125   }
126   if (cursor_ < cache_->cache_buffer_.count) {
127     *slice = grpc_slice_ref_internal(cache_->cache_buffer_.slices[cursor_]);
128     ++cursor_;
129     offset_ += GRPC_SLICE_LENGTH(*slice);
130     return GRPC_ERROR_NONE;
131   }
132   GPR_ASSERT(cache_->underlying_stream_ != nullptr);
133   grpc_error* error = cache_->underlying_stream_->Pull(slice);
134   if (error == GRPC_ERROR_NONE) {
135     grpc_slice_buffer_add(&cache_->cache_buffer_,
136                           grpc_slice_ref_internal(*slice));
137     ++cursor_;
138     offset_ += GRPC_SLICE_LENGTH(*slice);
139     // Orphan the underlying stream if it's been drained.
140     if (offset_ == cache_->underlying_stream_->length()) {
141       cache_->underlying_stream_.reset();
142     }
143   }
144   return error;
145 }
146 
Shutdown(grpc_error * error)147 void ByteStreamCache::CachingByteStream::Shutdown(grpc_error* error) {
148   GRPC_ERROR_UNREF(shutdown_error_);
149   shutdown_error_ = GRPC_ERROR_REF(error);
150   if (cache_->underlying_stream_ != nullptr) {
151     cache_->underlying_stream_->Shutdown(error);
152   }
153 }
154 
Reset()155 void ByteStreamCache::CachingByteStream::Reset() {
156   cursor_ = 0;
157   offset_ = 0;
158 }
159 
160 }  // namespace grpc_core
161