Skip to content
This repository was archived by the owner on Apr 8, 2025. It is now read-only.

Commit 73356c7

Browse files
Jinming-HuJinming-Hu
committed
Catch unobserved exception
1 parent 7de3384 commit 73356c7

File tree

4 files changed

+114
-21
lines changed

4 files changed

+114
-21
lines changed

‎Microsoft.WindowsAzure.Storage/src/cloud_append_blob.cpp

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -175,9 +175,27 @@ namespace azure { namespace storage {
175175

176176
return open_write_async_impl(create_new, condition, modified_options, context, cancellation_token, false, timer_handler).then([source, length, cancellation_token, timer_handler](concurrency::streams::ostream blob_stream) -> pplx::task<void>
177177
{
178-
return core::stream_copy_async(source, blob_stream, length, std::numeric_limits<utility::size64_t>::max(), cancellation_token, timer_handler).then([blob_stream](utility::size64_t) -> pplx::task<void>
178+
return core::stream_copy_async(source, blob_stream, length, std::numeric_limits<utility::size64_t>::max(), cancellation_token, timer_handler).then([blob_stream](pplx::task<utility::size64_t> copy_task) -> pplx::task<void>
179179
{
180-
return blob_stream.close();
180+
return blob_stream.close().then([copy_task](pplx::task<void> close_task)
181+
{
182+
try
183+
{
184+
copy_task.wait();
185+
}
186+
catch (const std::exception&)
187+
{
188+
try
189+
{
190+
close_task.wait();
191+
}
192+
catch (...)
193+
{
194+
}
195+
throw;
196+
}
197+
close_task.wait();
198+
});
181199
});
182200
});
183201
}

‎Microsoft.WindowsAzure.Storage/src/cloud_blob.cpp

Lines changed: 54 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -736,42 +736,57 @@ namespace azure { namespace storage {
736736

737737
auto smallest_offset = std::make_shared<utility::size64_t>(target_offset);
738738
auto condition_variable = std::make_shared<std::condition_variable>();
739-
std::mutex condition_variable_mutex;
739+
std::mutex condition_variable_mutex;
740+
std::vector<pplx::task<void>> parallel_tasks;
740741
for (utility::size64_t current_offset = target_offset; current_offset < target_offset + target_length; current_offset += protocol::transactional_md5_block_size)
741742
{
742743
utility::size64_t current_length = protocol::transactional_md5_block_size;
743744
if (current_offset + current_length > target_offset + target_length)
744745
{
745746
current_length = target_offset + target_length - current_offset;
746747
}
747-
semaphore->lock_async().then([instance, &mutex, semaphore, condition_variable, &condition_variable_mutex, &writer, offset, target, smallest_offset, current_offset, current_length, modified_condition, options, context, timer_handler]()
748+
auto parallel_task = semaphore->lock_async().then([instance, &mutex, semaphore, condition_variable, &condition_variable_mutex, &writer, offset, target, smallest_offset, current_offset, current_length, modified_condition, options, context, timer_handler]()
748749
{
750+
auto sem_unlocker = std::make_shared<std::unique_lock<core::async_semaphore>>(*semaphore, std::adopt_lock);
751+
749752
concurrency::streams::container_buffer<std::vector<uint8_t>> buffer;
750753
auto segment_ostream = buffer.create_ostream();
751754
// if transaction MD5 is enabled, it will be checked inside each download_single_range_to_stream_async.
752-
instance->download_single_range_to_stream_async(segment_ostream, current_offset, current_length, modified_condition, options, context, false, timer_handler->get_cancellation_token(), timer_handler)
753-
.then([buffer, segment_ostream, semaphore, condition_variable, &condition_variable_mutex, smallest_offset, offset, current_offset, current_length, &mutex, target, &writer, options](pplx::task<void> download_task)
755+
return instance->download_single_range_to_stream_async(segment_ostream, current_offset, current_length, modified_condition, options, context, false, timer_handler->get_cancellation_token(), timer_handler)
756+
.then([buffer, segment_ostream, semaphore, sem_unlocker, condition_variable, &condition_variable_mutex, smallest_offset, offset, current_offset, current_length, &mutex, target, &writer, options](pplx::task<void> download_task)
754757
{
755758
segment_ostream.close().then([download_task](pplx::task<void> close_task)
756759
{
757-
download_task.wait();
760+
try
761+
{
762+
download_task.wait();
763+
}
764+
catch (const std::exception&)
765+
{
766+
try
767+
{
768+
close_task.wait();
769+
}
770+
catch (...)
771+
{
772+
}
773+
throw;
774+
}
758775
close_task.wait();
759776
}).wait();
760777

761-
// status of current semaphore.
762-
bool released = false;
763778
// target stream is seekable, could write to target stream once the download finished.
764779
if (target.can_seek())
765780
{
766781
pplx::extensibility::scoped_rw_lock_t guard(mutex);
767782
target.streambuf().seekpos(current_offset - offset, std::ios_base::out);
768783
target.streambuf().putn_nocopy(buffer.collection().data(), buffer.collection().size()).wait();
769784
*smallest_offset += protocol::transactional_md5_block_size;
770-
released = true;
771-
semaphore->unlock();
772785
}
773786
else
774787
{
788+
// status of current semaphore.
789+
bool released = false;
775790
{
776791
pplx::extensibility::scoped_rw_lock_t guard(mutex);
777792
if (*smallest_offset == current_offset)
@@ -781,7 +796,7 @@ namespace azure { namespace storage {
781796
*smallest_offset += protocol::transactional_md5_block_size;
782797
condition_variable->notify_all();
783798
released = true;
784-
semaphore->unlock();
799+
sem_unlocker->unlock();
785800
}
786801
}
787802
if (!released)
@@ -790,7 +805,7 @@ namespace azure { namespace storage {
790805
if (writer < options.parallelism_factor())
791806
{
792807
released = true;
793-
semaphore->unlock();
808+
sem_unlocker->unlock();
794809
}
795810
std::unique_lock<std::mutex> locker(condition_variable_mutex);
796811
condition_variable->wait(locker, [smallest_offset, current_offset, &mutex]()
@@ -813,15 +828,13 @@ namespace azure { namespace storage {
813828
}
814829
condition_variable->notify_all();
815830
pplx::details::atomic_decrement(writer);
816-
if (!released)
817-
{
818-
semaphore->unlock();
819-
}
820831
}
821832
}
822833
});
823834
});
835+
parallel_tasks.emplace_back(std::move(parallel_task));
824836
}
837+
// Code below is nonsense, becasuse exceptions won't be thrown from wait_all_async.
825838
// If the cancellation token is canceled, the lock will be in lock status when the exception is thrown, so need to unlock it in case it blocks other async processes
826839
try
827840
{
@@ -835,6 +848,32 @@ namespace azure { namespace storage {
835848
}
836849
throw ex;
837850
}
851+
852+
pplx::when_all(parallel_tasks.begin(), parallel_tasks.end()).then([parallel_tasks](pplx::task<void> wait_all_task)
853+
{
854+
try
855+
{
856+
wait_all_task.wait();
857+
}
858+
catch (const std::exception&)
859+
{
860+
std::for_each(parallel_tasks.begin(), parallel_tasks.end(), [](pplx::task<void> task)
861+
{
862+
task.then([](pplx::task<void> t)
863+
{
864+
try
865+
{
866+
t.wait();
867+
}
868+
catch (...)
869+
{
870+
}
871+
});
872+
});
873+
throw;
874+
}
875+
}).wait();
876+
838877
std::unique_lock<std::mutex> locker(condition_variable_mutex);
839878
condition_variable->wait(locker, [smallest_offset, &mutex, target_offset, target_length]()
840879
{

‎Microsoft.WindowsAzure.Storage/src/cloud_block_blob.cpp

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -261,9 +261,27 @@ namespace azure { namespace storage {
261261

262262
return open_write_async_impl(condition, modified_options, context, timer_handler->get_cancellation_token(), false, timer_handler).then([source, length, timer_handler](concurrency::streams::ostream blob_stream) -> pplx::task<void>
263263
{
264-
return core::stream_copy_async(source, blob_stream, length, std::numeric_limits<utility::size64_t>::max(), timer_handler->get_cancellation_token(), timer_handler).then([blob_stream, timer_handler](utility::size64_t)->pplx::task<void>
264+
return core::stream_copy_async(source, blob_stream, length, std::numeric_limits<utility::size64_t>::max(), timer_handler->get_cancellation_token(), timer_handler).then([blob_stream, timer_handler](pplx::task<utility::size64_t> copy_task)->pplx::task<void>
265265
{
266-
return blob_stream.close().then([timer_handler/*timer_handler MUST be captured*/]() {});
266+
return blob_stream.close().then([timer_handler, copy_task](pplx::task<void> close_task)
267+
{
268+
try
269+
{
270+
copy_task.wait();
271+
}
272+
catch (const std::exception&)
273+
{
274+
try
275+
{
276+
close_task.wait();
277+
}
278+
catch (...)
279+
{
280+
}
281+
throw;
282+
}
283+
close_task.wait();
284+
});
267285
});
268286
});
269287
}

‎Microsoft.WindowsAzure.Storage/src/cloud_page_blob.cpp

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -141,9 +141,27 @@ namespace azure { namespace storage {
141141

142142
return open_write_async_impl(length, sequence_number, condition, modified_options, context, timer_handler->get_cancellation_token(), false, timer_handler).then([source, length, timer_handler, options](concurrency::streams::ostream blob_stream) -> pplx::task<void>
143143
{
144-
return core::stream_copy_async(source, blob_stream, length, std::numeric_limits<utility::size64_t>::max(), timer_handler->get_cancellation_token(), timer_handler).then([blob_stream, timer_handler, options] (utility::size64_t) -> pplx::task<void>
144+
return core::stream_copy_async(source, blob_stream, length, std::numeric_limits<utility::size64_t>::max(), timer_handler->get_cancellation_token(), timer_handler).then([blob_stream, timer_handler, options] (pplx::task<utility::size64_t> copy_task) -> pplx::task<void>
145145
{
146-
return blob_stream.close().then([timer_handler/*timer_handler MUST be captured*/]() {});
146+
return blob_stream.close().then([timer_handler, copy_task](pplx::task<void> close_task)
147+
{
148+
try
149+
{
150+
copy_task.wait();
151+
}
152+
catch (const std::exception&)
153+
{
154+
try
155+
{
156+
close_task.wait();
157+
}
158+
catch (...)
159+
{
160+
}
161+
throw;
162+
}
163+
close_task.wait();
164+
});
147165
});
148166
});
149167
}

0 commit comments

Comments
 (0)