Deprecating stream buffer putn function, adding putn_nocopy. This is really is the best choice to completely remove the unnecessary copy in file streams.

Author: stgates

Release/include/cpprest/astreambuf.h

@@ -197,10 +197,19 @@ namespace streams
         /// <returns>A <c>task</c> that holds the number of characters actually written, either 'count' or 0.</returns>
         virtual pplx::task<size_t> putn(const _CharType *ptr, size_t count) = 0;
 
+        /// <summary>
+        /// Writes a number of characters to the stream. Note: callers must make sure the data to be written is valid until
+        /// the returned task completes.
+        /// </summary>
+        /// <param name="ptr">A pointer to the block of data to be written.</param>
+        /// <param name="count">The number of characters to write.</param>
+        /// <returns>A <c>task</c> that holds the number of characters actually written, either 'count' or 0.</returns>
+        virtual pplx::task<size_t> putn_nocopy(const _CharType *ptr, size_t count) = 0;
+
         /// <summary>
         /// Reads a single character from the stream and advances the read position.
         /// </summary>
-            /// <returns>A <c>task</c> that holds the value of the character. This value is EOF if the read fails.</returns>
+        /// <returns>A <c>task</c> that holds the value of the character. This value is EOF if the read fails.</returns>
         virtual pplx::task<int_type> bumpc() = 0;
 
         /// <summary>
@@ -226,7 +235,7 @@ namespace streams
         /// <summary>
         /// Advances the read position, then returns the next character without advancing again.
         /// </summary>
-            /// <returns>A <c>task</c> that holds the value of the character. This value is EOF if the read fails.</returns>
+        /// <returns>A <c>task</c> that holds the value of the character. This value is EOF if the read fails.</returns>
         virtual pplx::task<int_type> nextc() = 0;
 
         /// <summary>
@@ -450,7 +459,27 @@ namespace streams
         /// <param name="ptr">A pointer to the block of data to be written.</param>
         /// <param name="count">The number of characters to write.</param>
         /// <returns>The number of characters actually written, either 'count' or 0.</returns>
+        CASABLANCA_DEPRECATED("This API in some cases performs a copy. It is deprecated and will be removed in a future release. Use putn_nocopy instead.")
         virtual pplx::task<size_t> putn(const _CharType *ptr, size_t count)
+        {
+            if (!can_write())
+                return create_exception_checked_value_task<size_t>(0);
+            if (count == 0)
+                return pplx::task_from_result<size_t>(0);
+
+            return create_exception_checked_task<size_t>(_putn(ptr, count, true), [](size_t) {
+                return false; // no EOF for write
+            });
+        }
+
+        /// <summary>
+        /// Writes a number of characters to the stream. Note: callers must make sure the data to be written is valid until
+        /// the returned task completes.
+        /// </summary>
+        /// <param name="ptr">A pointer to the block of data to be written.</param>
+        /// <param name="count">The number of characters to write.</param>
+        /// <returns>A <c>task</c> that holds the number of characters actually written, either 'count' or 0.</returns>
+        virtual pplx::task<size_t> putn_nocopy(const _CharType *ptr, size_t count)
         {
             if (!can_write())
                 return create_exception_checked_value_task<size_t>(0);
@@ -655,7 +684,15 @@ namespace streams
         virtual void release(_Out_writes_(count) _CharType *ptr, _In_ size_t count) = 0;
     protected:
         virtual pplx::task<int_type> _putc(_CharType ch) = 0;
+
+        // This API is only needed for file streams and until we remove the deprecated stream buffer putn overload.
+        virtual pplx::task<size_t> _putn(const _CharType *ptr, size_t count, bool)
+        {
+            // Default to no copy, only the file streams API overloads and performs a copy.
+            return _putn(ptr, count);
+        }
         virtual pplx::task<size_t> _putn(const _CharType *ptr, size_t count) = 0;
+
         virtual pplx::task<int_type> _bumpc() = 0;
         virtual int_type _sbumpc() = 0;
         virtual pplx::task<int_type> _getc() = 0;
@@ -866,7 +903,7 @@ namespace streams
         /// </summary>
         virtual ~streambuf() { }
 
-        std::shared_ptr<details::basic_streambuf<_CharType>> get_base() const
+        const std::shared_ptr<details::basic_streambuf<_CharType>> & get_base() const
         {
             if (!m_buffer)
             {
@@ -1031,11 +1068,24 @@ namespace streams
         /// <param name="ptr">A pointer to the block of data to be written.</param>
         /// <param name="count">The number of characters to write.</param>
         /// <returns>The number of characters actually written, either 'count' or 0.</returns>
+        CASABLANCA_DEPRECATED("This API in some cases performs a copy. It is deprecated and will be removed in a future release. Use putn_nocopy instead.")
         virtual pplx::task<size_t> putn(const _CharType *ptr, size_t count)
         {
             return get_base()->putn(ptr, count);
         }
 
+        /// <summary>
+        /// Writes a number of characters to the stream. Note: callers must make sure the data to be written is valid until
+        /// the returned task completes.
+        /// </summary>
+        /// <param name="ptr">A pointer to the block of data to be written.</param>
+        /// <param name="count">The number of characters to write.</param>
+        /// <returns>The number of characters actually written, either 'count' or 0.</returns>
+        virtual pplx::task<size_t> putn_nocopy(const _CharType *ptr, size_t count)
+        {
+            return get_base()->putn_nocopy(ptr, count);
+        }
+
         /// <summary>
         /// Reads a single character from the stream and advances the read position.
         /// </summary>

Release/include/cpprest/filestream.h

@@ -368,6 +368,24 @@ namespace details {
             return pplx::create_task(result_tce);
         }
 
+        // Temporarily needed until the deprecated putn is removed.
+        virtual pplx::task<size_t> _putn(const _CharType *ptr, size_t count, bool copy)
+        {
+            if (copy)
+            {
+                auto sharedData = std::shared_ptr<_CharType>(new _CharType[count], std::default_delete<_CharType []>());
+                memcpy(sharedData.get(), ptr, count * sizeof(_CharType));
+                return _putn(ptr, count).then([sharedData](size_t size)
+                {
+                    return size;
+                });
+            }
+            else
+            {
+                return _putn(ptr, count);
+            }
+        }
+
         /// <summary>
         /// Reads a single byte from the stream and advance the read position.
         /// </summary>
@@ -741,7 +759,7 @@ namespace details {
         class _filestream_callback_open : public details::_filestream_callback
         {
         public:
-            _filestream_callback_open(pplx::task_completion_event<std::shared_ptr<basic_streambuf<_CharType>>> op) : m_op(op) { }
+            _filestream_callback_open(const pplx::task_completion_event<std::shared_ptr<basic_streambuf<_CharType>>> &op) : m_op(op) { }
 
             virtual void on_opened(_In_ _file_info *info)
             {
@@ -762,7 +780,7 @@ namespace details {
         class _filestream_callback_close : public details::_filestream_callback
         {
         public:
-            _filestream_callback_close(pplx::task_completion_event<void> op) : m_op(op) { }
+            _filestream_callback_close(const pplx::task_completion_event<void> &op) : m_op(op) { }
 
             virtual void on_closed()
             {
@@ -784,7 +802,7 @@ namespace details {
         class _filestream_callback_write : public details::_filestream_callback
         {
         public:
-            _filestream_callback_write(_In_ _file_info *info, pplx::task_completion_event<ResultType> op) : m_info(info), m_op(op) { }
+            _filestream_callback_write(_In_ _file_info *info, const pplx::task_completion_event<ResultType> &op) : m_info(info), m_op(op) { }
 
             virtual void on_completed(size_t result)
             {
@@ -806,7 +824,7 @@ namespace details {
         class _filestream_callback_write_b : public details::_filestream_callback
         {
         public:
-            _filestream_callback_write_b(_In_ _file_info *info, pplx::task_completion_event<void> op) : m_info(info), m_op(op) { }
+            _filestream_callback_write_b(_In_ _file_info *info, const pplx::task_completion_event<void> &op) : m_info(info), m_op(op) { }
 
             virtual void on_completed(size_t)
             {
@@ -829,7 +847,7 @@ namespace details {
         class _filestream_callback_read : public details::_filestream_callback
         {
         public:
-            _filestream_callback_read(_In_ _file_info *info, pplx::task_completion_event<size_t> op) : m_info(info), m_op(op) { }
+            _filestream_callback_read(_In_ _file_info *info, const pplx::task_completion_event<size_t> &op) : m_info(info), m_op(op) { }
 
             virtual void on_completed(size_t result)
             {
@@ -853,7 +871,7 @@ namespace details {
         class _filestream_callback_bumpc : public details::_filestream_callback
         {
         public:
-            _filestream_callback_bumpc(_In_ _file_info *info, pplx::task_completion_event<int_type> op) : m_ch(0), m_info(info), m_op(op) { }
+            _filestream_callback_bumpc(_In_ _file_info *info, const pplx::task_completion_event<int_type> &op) : m_ch(0), m_info(info), m_op(op) { }
 
             virtual void on_completed(size_t result)
             {
@@ -885,7 +903,7 @@ namespace details {
         class _filestream_callback_getc : public details::_filestream_callback
         {
         public:
-            _filestream_callback_getc(_In_ _file_info *info, pplx::task_completion_event<int_type> op) : m_ch(0), m_info(info), m_op(op) { }
+            _filestream_callback_getc(_In_ _file_info *info, const pplx::task_completion_event<int_type> &op) : m_ch(0), m_info(info), m_op(op) { }
 
             virtual void on_completed(size_t result)
             {

Release/include/cpprest/interopstream.h

@@ -309,8 +309,8 @@ namespace Concurrency { namespace streams {
         {
             try
             {
-            return m_buffer.putn(ptr, (size_t)count).get();
-        }
+                return m_buffer.putn_nocopy(ptr, static_cast<size_t>(count)).get();
+            }
             catch(...)
             {
                 return 0;

Release/include/cpprest/streams.h

@@ -196,7 +196,7 @@ namespace Concurrency { namespace streams
             if ( !_verify_and_return_task(details::_out_stream_msg, result) ) return result;
 
             auto copy = std::make_shared<T>(std::move(value));
-            return helper()->m_buffer.putn((CharType*)copy.get(), sizeof(T)).then([copy](pplx::task<size_t> op) -> size_t { return op.get(); });
+            return helper()->m_buffer.putn_nocopy((CharType*)copy.get(), sizeof(T)).then([copy](pplx::task<size_t> op) -> size_t { return op.get(); });
         }
 
         /// <summary>
@@ -242,7 +242,7 @@ namespace Concurrency { namespace streams
                             s.release(data,op.get());
                             return op;
                         };
-                    return buffer.putn(data, count).then(post_write);
+                    return buffer.putn_nocopy(data, count).then(post_write);
                 }
                 else
                 {
@@ -263,7 +263,7 @@ namespace Concurrency { namespace streams
                         [buf,post_write,buffer](pplx::task<size_t> op) -> pplx::task<size_t>
                         {
                             auto b = buffer;
-                            return b.putn(buf.get(), op.get()).then(post_write);
+                            return b.putn_nocopy(buf.get(), op.get()).then(post_write);
                         };
 
                     return source.getn(buf.get(), count).then(post_read);
@@ -279,9 +279,16 @@ namespace Concurrency { namespace streams
         {
             pplx::task<size_t> result;
             if ( !_verify_and_return_task(details::_out_stream_msg, result) ) return result;
-            return (str.size() == 0)
-                ? pplx::task_from_result<size_t>((0))
-                : helper()->m_buffer.putn(str.c_str(), str.size());
+
+            if (str.empty())
+            {
+                return pplx::task_from_result<size_t>(0);
+            }
+            else
+            {
+                auto sharedStr = std::make_shared<std::basic_string<CharType>>(str);
+                return helper()->m_buffer.putn_nocopy(sharedStr->c_str(), sharedStr->size()).then([sharedStr](size_t size) { return size; });
+            }
         }
 
         /// <summary>
@@ -713,7 +720,7 @@ namespace Concurrency { namespace streams
                             b.release(data, op.get());
                             return op;
                         };
-                    return target.putn(data, count).then(post_write);
+                    return target.putn_nocopy(data, count).then(post_write);
                 }
                 else
                 {
@@ -734,7 +741,7 @@ namespace Concurrency { namespace streams
                         [buf,target,post_write](pplx::task<size_t> op) -> pplx::task<size_t>
                         {
                             auto trg = target;
-                            return trg.putn(buf.get(), op.get()).then(post_write);
+                            return trg.putn_nocopy(buf.get(), op.get()).then(post_write);
                         };
 
                     return helper()->m_buffer.getn(buf.get(), count).then(post_read);
@@ -776,7 +783,7 @@ namespace Concurrency { namespace streams
 
             auto flush = [=]() mutable
             {
-                return target.putn(_locals->outbuf, _locals->write_pos).then([=](size_t wrote) mutable
+                return target.putn_nocopy(_locals->outbuf, _locals->write_pos).then([=](size_t wrote) mutable
                 {
                     _locals->total += wrote;
                     _locals->write_pos = 0;
@@ -848,7 +855,7 @@ namespace Concurrency { namespace streams
 
             auto flush = [=]() mutable
             {
-                return target.putn(_locals->outbuf, _locals->write_pos).then([=](size_t wrote) mutable
+                return target.putn_nocopy(_locals->outbuf, _locals->write_pos).then([=](size_t wrote) mutable
                 {
                     _locals->total += wrote;
                     _locals->write_pos = 0;
@@ -963,7 +970,7 @@ namespace Concurrency { namespace streams
                         return pplx::task_from_result(false);
 
                     // Must be nested to capture rd
-                    return target.putn(l_locals->outbuf, rd).then([target, l_locals, rd](size_t wr) mutable -> pplx::task<bool>
+                    return target.putn_nocopy(l_locals->outbuf, rd).then([target, l_locals, rd](size_t wr) mutable -> pplx::task<bool>
                     {
                         l_locals->total += wr;
 
@@ -1728,7 +1735,7 @@ class type_parser<char,std::basic_string<wchar_t>> : public _type_parser_base<ch
     }
 
 private:
-    static bool _accept_char(std::shared_ptr<std::basic_string<char>> state, int_type ch)
+    static bool _accept_char(const std::shared_ptr<std::basic_string<char>> &state, int_type ch)
     {
         if ( ch == concurrency::streams::char_traits<char>::eof() || isspace(ch)) return false;
         state->push_back(char(ch));
@@ -1750,7 +1757,7 @@ class type_parser<signed char,std::basic_string<wchar_t>> : public _type_parser_
     }
 
 private:
-    static bool _accept_char(std::shared_ptr<std::basic_string<char>> state, int_type ch)
+    static bool _accept_char(const std::shared_ptr<std::basic_string<char>> &state, int_type ch)
     {
         if ( ch == concurrency::streams::char_traits<char>::eof() || isspace(ch)) return false;
         state->push_back(char(ch));
@@ -1772,7 +1779,7 @@ class type_parser<unsigned char,std::basic_string<wchar_t>> : public _type_parse
     }
 
 private:
-    static bool _accept_char(std::shared_ptr<std::basic_string<char>> state, int_type ch)
+    static bool _accept_char(const std::shared_ptr<std::basic_string<char>> &state, int_type ch)
     {
         if ( ch == concurrency::streams::char_traits<char>::eof() || isspace(ch)) return false;
         state->push_back(char(ch));

Release/src/http/client/http_client_asio.cpp

@@ -968,7 +968,7 @@ class asio_context : public request_context, public std::enable_shared_from_this
             {
                 auto writeBuffer = _get_writebuffer();
                 const auto this_request = shared_from_this();
-                writeBuffer.putn(boost::asio::buffer_cast<const uint8_t *>(m_body_buf.data()), to_read).then([this_request, to_read](pplx::task<size_t> op)
+                writeBuffer.putn_nocopy(boost::asio::buffer_cast<const uint8_t *>(m_body_buf.data()), to_read).then([this_request, to_read](pplx::task<size_t> op)
                 {
                     try
                     {
@@ -1026,7 +1026,7 @@ class asio_context : public request_context, public std::enable_shared_from_this
         {
             // more data need to be read
             const auto this_request = shared_from_this();
-            writeBuffer.putn(boost::asio::buffer_cast<const uint8_t *>(m_body_buf.data()),
+            writeBuffer.putn_nocopy(boost::asio::buffer_cast<const uint8_t *>(m_body_buf.data()),
                              static_cast<size_t>(std::min(static_cast<uint64_t>(m_body_buf.size()), m_content_length - m_downloaded)))
             .then([this_request](pplx::task<size_t> op)
             {

Release/src/http/client/http_client_winhttp.cpp

@@ -739,7 +739,7 @@ class winhttp_client : public _http_client_communicator
                 {
                     // We have raw memory here writing to a memory stream so it is safe to wait
                     // since it will always be non-blocking.
-                    p_request_context->m_readBufferCopy->putn(&p_request_context->m_body_data.get()[http::details::chunked_encoding::data_offset], bytes_read).wait();
+                    p_request_context->m_readBufferCopy->putn_nocopy(&p_request_context->m_body_data.get()[http::details::chunked_encoding::data_offset], bytes_read).wait();
                 }
             }
             catch (...)
@@ -1230,7 +1230,7 @@ class winhttp_client : public _http_client_communicator
                     }
                     else
                     {
-                        writebuf.putn(p_request_context->m_body_data.get(), bytesRead).then(
+                        writebuf.putn_nocopy(p_request_context->m_body_data.get(), bytesRead).then(
                             [hRequestHandle, p_request_context, bytesRead] (pplx::task<size_t> op)
                         {
                             size_t written = 0;

Release/src/http/client/http_client_winrt.cpp

@@ -307,7 +307,7 @@ class IResponseStream
         try
         {
             auto buffer = context->_get_writebuffer();
-            const size_t count = buffer.putn(reinterpret_cast<const uint8_t *>(pv), static_cast<size_t>(cb)).get();
+            const size_t count = buffer.putn_nocopy(reinterpret_cast<const uint8_t *>(pv), static_cast<size_t>(cb)).get();
 
             _ASSERTE(count != static_cast<size_t>(-1));
             _ASSERTE(count <= static_cast<size_t>(ULONG_MAX));