1 /*
2 * Copyright (C) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "src_bytebuffer_impl.h"
17 #include <mutex>
18 #include "gst_shmem_memory.h"
19 #include "media_log.h"
20 #include "scope_guard.h"
21
22 namespace {
23 constexpr OHOS::HiviewDFX::HiLogLabel LABEL = {LOG_CORE, LOG_DOMAIN, "SrcBytebufferImpl"};
24 constexpr uint32_t DEFAULT_BUFFER_NUM = 10;
25 }
26
27 namespace OHOS {
28 namespace Media {
SrcBytebufferImpl()29 SrcBytebufferImpl::SrcBytebufferImpl()
30 {
31 }
32
~SrcBytebufferImpl()33 SrcBytebufferImpl::~SrcBytebufferImpl()
34 {
35 bufferList_.clear();
36 if (src_ != nullptr) {
37 gst_object_unref(src_);
38 src_ = nullptr;
39 }
40 if (caps_ != nullptr) {
41 gst_caps_unref(caps_);
42 caps_ = nullptr;
43 }
44 }
45
Init()46 int32_t SrcBytebufferImpl::Init()
47 {
48 src_ = GST_ELEMENT_CAST(gst_object_ref(gst_element_factory_make("shmemsrc", "src")));
49 CHECK_AND_RETURN_RET_LOG(src_ != nullptr, MSERR_UNKNOWN, "Failed to gst_element_factory_make");
50 return MSERR_OK;
51 }
52
Configure(std::shared_ptr<ProcessorConfig> config)53 int32_t SrcBytebufferImpl::Configure(std::shared_ptr<ProcessorConfig> config)
54 {
55 CHECK_AND_RETURN_RET(src_ != nullptr, MSERR_UNKNOWN);
56
57 g_object_set(G_OBJECT(src_), "buffer-num", DEFAULT_BUFFER_NUM, nullptr);
58 g_object_set(G_OBJECT(src_), "buffer-size", config->bufferSize_, nullptr);
59 g_object_set(G_OBJECT(src_), "caps", config->caps_, nullptr);
60 gst_mem_pool_src_set_callback(GST_MEM_POOL_SRC(src_), BufferAvailable, this, nullptr);
61
62 needCodecData_ = config->needCodecData_;
63 if (needCodecData_) {
64 caps_ = config->caps_;
65 gst_caps_ref(caps_);
66 }
67 return MSERR_OK;
68 }
69
Start()70 int32_t SrcBytebufferImpl::Start()
71 {
72 std::unique_lock<std::mutex> lock(mutex_);
73 start_ = true;
74 return MSERR_OK;
75 }
76
Stop()77 int32_t SrcBytebufferImpl::Stop()
78 {
79 std::unique_lock<std::mutex> lock(mutex_);
80 start_ = false;
81 return MSERR_OK;
82 }
83
Flush()84 int32_t SrcBytebufferImpl::Flush()
85 {
86 std::unique_lock<std::mutex> lock(mutex_);
87 for (auto it = bufferList_.begin(); it != bufferList_.end(); it++) {
88 if ((*it)->owner_ != BufferWrapper::DOWNSTREAM) {
89 (*it)->owner_ = BufferWrapper::DOWNSTREAM;
90 if ((*it)->gstBuffer_ != nullptr) {
91 gst_buffer_unref((*it)->gstBuffer_);
92 (*it)->gstBuffer_ = nullptr;
93 }
94 }
95 }
96 return MSERR_OK;
97 }
98
Needflush()99 bool SrcBytebufferImpl::Needflush()
100 {
101 std::unique_lock<std::mutex> lock(mutex_);
102 return !bufferList_.empty();
103 }
104
GetInputBuffer(uint32_t index)105 std::shared_ptr<AVSharedMemory> SrcBytebufferImpl::GetInputBuffer(uint32_t index)
106 {
107 std::unique_lock<std::mutex> lock(mutex_);
108 CHECK_AND_RETURN_RET(index <= bufferList_.size(), nullptr);
109 CHECK_AND_RETURN_RET(bufferList_[index]->owner_ == BufferWrapper::SERVER, nullptr);
110
111 GstMemory *memory = gst_buffer_peek_memory(bufferList_[index]->gstBuffer_, 0);
112 CHECK_AND_RETURN_RET(memory != nullptr, nullptr);
113
114 GstShMemMemory *shmem = reinterpret_cast<GstShMemMemory *>(memory);
115 bufferList_[index]->owner_ = BufferWrapper::APP;
116
117 return shmem->mem;
118 }
119
QueueInputBuffer(uint32_t index,AVCodecBufferInfo info,AVCodecBufferFlag flag)120 int32_t SrcBytebufferImpl::QueueInputBuffer(uint32_t index, AVCodecBufferInfo info, AVCodecBufferFlag flag)
121 {
122 std::unique_lock<std::mutex> lock(mutex_);
123 CHECK_AND_RETURN_RET(index <= bufferList_.size(), MSERR_INVALID_VAL);
124
125 auto &bufWrapper = bufferList_[index];
126 CHECK_AND_RETURN_RET(bufWrapper->owner_ == BufferWrapper::APP, MSERR_INVALID_OPERATION);
127 CHECK_AND_RETURN_RET(bufWrapper->gstBuffer_ != nullptr, MSERR_INVALID_OPERATION);
128 gst_buffer_resize(bufWrapper->gstBuffer_, info.offset, info.size);
129
130 if (needCodecData_) {
131 if (HandleCodecBuffer(index, info, flag) == MSERR_OK) {
132 needCodecData_ = false;
133 bufWrapper->owner_ = BufferWrapper::SERVER;
134 auto obs = obs_.lock();
135 CHECK_AND_RETURN_RET(obs != nullptr, MSERR_UNKNOWN);
136 obs->OnInputBufferAvailable(index);
137 MEDIA_LOGD("OnInputBufferAvailable, index:%{public}u", index);
138 return MSERR_OK;
139 }
140 return MSERR_UNKNOWN;
141 }
142
143 uint8_t *address = bufWrapper->mem_->GetBase();
144 CHECK_AND_RETURN_RET(address != nullptr, MSERR_UNKNOWN);
145 CHECK_AND_RETURN_RET((info.offset + info.size) <= bufWrapper->mem_->GetSize(), MSERR_INVALID_VAL);
146
147 constexpr int32_t usToNs = 1000;
148 if (info.presentationTimeUs < 0) {
149 MEDIA_LOGE("Invalid pts: < 0, use 0 as default");
150 GST_BUFFER_PTS(bufWrapper->gstBuffer_) = 0;
151 } else if ((INT64_MAX / usToNs) <= info.presentationTimeUs) {
152 MEDIA_LOGE("Invalid pts: too big, use 0 as default");
153 GST_BUFFER_PTS(bufWrapper->gstBuffer_) = 0;
154 } else {
155 GST_BUFFER_PTS(bufWrapper->gstBuffer_) = static_cast<uint64_t>(info.presentationTimeUs * usToNs);
156 }
157
158 CHECK_AND_RETURN_RET(src_ != nullptr, MSERR_UNKNOWN);
159 (void)gst_mem_pool_src_push_buffer(GST_MEM_POOL_SRC(src_), bufWrapper->gstBuffer_);
160 bufWrapper->owner_ = BufferWrapper::DOWNSTREAM;
161 bufWrapper->gstBuffer_ = nullptr; // src elem take ownership of this buffer.
162
163 MEDIA_LOGD("QueueInputBuffer, index = %{public}u", index);
164 return MSERR_OK;
165 }
166
SetCallback(const std::weak_ptr<IAVCodecEngineObs> & obs)167 int32_t SrcBytebufferImpl::SetCallback(const std::weak_ptr<IAVCodecEngineObs> &obs)
168 {
169 std::unique_lock<std::mutex> lock(mutex_);
170 obs_ = obs;
171 return MSERR_OK;
172 }
173
HandleCodecBuffer(uint32_t index,AVCodecBufferInfo info,AVCodecBufferFlag flag)174 int32_t SrcBytebufferImpl::HandleCodecBuffer(uint32_t index, AVCodecBufferInfo info, AVCodecBufferFlag flag)
175 {
176 bool hasCodecFlag = static_cast<uint32_t>(flag) & static_cast<uint32_t>(AVCODEC_BUFFER_FLAG_CODEC_DATA);
177 CHECK_AND_RETURN_RET_LOG(hasCodecFlag == true, MSERR_INVALID_VAL, "First buffer must be codec buffer");
178
179 uint8_t *address = bufferList_[index]->mem_->GetBase();
180 CHECK_AND_RETURN_RET(address != nullptr, MSERR_UNKNOWN);
181 CHECK_AND_RETURN_RET((info.offset + info.size) <= bufferList_[index]->mem_->GetSize(), MSERR_INVALID_VAL);
182
183 GstBuffer *codecBuffer = gst_buffer_new_allocate(nullptr, info.size, nullptr);
184 CHECK_AND_RETURN_RET_LOG(codecBuffer != nullptr, MSERR_NO_MEMORY, "no memory");
185
186 ON_SCOPE_EXIT(0) { gst_buffer_unref(codecBuffer); };
187
188 gsize size = gst_buffer_fill(codecBuffer, 0, (char *)address + info.offset, info.size);
189 CHECK_AND_RETURN_RET(size == static_cast<gsize>(info.size), MSERR_UNKNOWN);
190
191 CHECK_AND_RETURN_RET(caps_ != nullptr, MSERR_UNKNOWN);
192 gst_caps_set_simple(caps_, "codec_data", GST_TYPE_BUFFER, codecBuffer, nullptr);
193
194 CHECK_AND_RETURN_RET(src_ != nullptr, MSERR_UNKNOWN);
195 g_object_set(G_OBJECT(src_), "caps", caps_, nullptr);
196
197 CHECK_AND_RETURN_RET(gst_base_src_set_caps(GST_BASE_SRC(src_), caps_) == TRUE, MSERR_UNKNOWN);
198 return MSERR_OK;
199 }
200
BufferAvailable(GstMemPoolSrc * memsrc,gpointer userdata)201 GstFlowReturn SrcBytebufferImpl::BufferAvailable(GstMemPoolSrc *memsrc, gpointer userdata)
202 {
203 CHECK_AND_RETURN_RET(memsrc != nullptr, GST_FLOW_ERROR);
204 CHECK_AND_RETURN_RET(userdata != nullptr, GST_FLOW_ERROR);
205
206 GstBuffer *buffer = gst_mem_pool_src_pull_buffer(memsrc);
207 CHECK_AND_RETURN_RET(buffer != nullptr, GST_FLOW_ERROR);
208
209 SrcBytebufferImpl *thiz = reinterpret_cast<SrcBytebufferImpl *>(userdata);
210 int32_t ret = thiz->HandleBufferAvailable(buffer);
211 CHECK_AND_RETURN_RET(ret == MSERR_OK, GST_FLOW_ERROR);
212
213 return GST_FLOW_OK;
214 }
215
HandleBufferAvailable(GstBuffer * buffer)216 int32_t SrcBytebufferImpl::HandleBufferAvailable(GstBuffer *buffer)
217 {
218 std::unique_lock<std::mutex> lock(mutex_);
219 ON_SCOPE_EXIT(0) { gst_buffer_unref(buffer); };
220 if (!start_) {
221 MEDIA_LOGD("Codec source is stop, unref available buffer");
222 return MSERR_OK;
223 }
224
225 GstMemory *memory = gst_buffer_peek_memory(buffer, 0);
226 CHECK_AND_RETURN_RET(memory != nullptr, MSERR_UNKNOWN);
227 GstShMemMemory *shmem = reinterpret_cast<GstShMemMemory *>(memory);
228 CHECK_AND_RETURN_RET(shmem->mem != nullptr, MSERR_UNKNOWN);
229
230 uint32_t index = 0;
231 int32_t ret = FindBufferIndex(index, shmem->mem);
232 CHECK_AND_RETURN_RET(ret == MSERR_OK, MSERR_UNKNOWN);
233
234 auto obs = obs_.lock();
235 CHECK_AND_RETURN_RET_LOG(obs != nullptr, MSERR_UNKNOWN, "obs is nullptr");
236 obs->OnInputBufferAvailable(index);
237
238 MEDIA_LOGD("OnInputBufferAvailable, index:%{public}u", index);
239 bufferList_[index]->owner_ = BufferWrapper::SERVER;
240 bufferList_[index]->gstBuffer_ = gst_buffer_ref(buffer);
241
242 return MSERR_OK;
243 }
244
FindBufferIndex(uint32_t & index,std::shared_ptr<AVSharedMemory> mem)245 int32_t SrcBytebufferImpl::FindBufferIndex(uint32_t &index, std::shared_ptr<AVSharedMemory> mem)
246 {
247 CHECK_AND_RETURN_RET(mem != nullptr, MSERR_UNKNOWN);
248
249 index = 0;
250 for (auto it = bufferList_.begin(); it != bufferList_.end(); it++) {
251 if ((*it) != nullptr && (*it)->mem_ == mem.get()) {
252 break;
253 }
254 index++;
255 }
256
257 if (index == bufferList_.size()) {
258 auto bufWrap = std::make_shared<BufferWrapper>(BufferWrapper::SERVER);
259 CHECK_AND_RETURN_RET(bufWrap != nullptr, MSERR_NO_MEMORY);
260 bufWrap->mem_ = mem.get();
261 bufferList_.push_back(bufWrap);
262 }
263
264 return MSERR_OK;
265 }
266 } // namespace Media
267 } // namespace OHOS
268