You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@trafficserver.apache.org by yu...@apache.org on 2014/03/12 04:12:46 UTC

[1/5] git commit: TS-2610: Add "client_protocol_stack"(%) field into LogFormat

Repository: trafficserver
Updated Branches:
  refs/heads/master 558345f21 -> f914a62fc


TS-2610: Add "client_protocol_stack"(%<cps>) field into LogFormat

The output of %<cps> field would be the conjunction of protocol names
in client protocol stack spliced with '+', such as "TLS+SPDY".

Signed-off-by: Yunkai Zhang <qi...@taobao.com>


Project: http://git-wip-us.apache.org/repos/asf/trafficserver/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafficserver/commit/f914a62f
Tree: http://git-wip-us.apache.org/repos/asf/trafficserver/tree/f914a62f
Diff: http://git-wip-us.apache.org/repos/asf/trafficserver/diff/f914a62f

Branch: refs/heads/master
Commit: f914a62fc3fc1f3d552b48dd7d6ca1283fb8c776
Parents: 25555f8
Author: Yunkai Zhang <qi...@taobao.com>
Authored: Mon Jan 6 16:59:19 2014 +0800
Committer: Yunkai Zhang <qi...@taobao.com>
Committed: Wed Mar 12 11:02:15 2014 +0800

----------------------------------------------------------------------
 doc/admin/event-logging-formats.en.rst |  4 +++
 proxy/logging/Log.cc                   | 20 ++++++++++++++
 proxy/logging/LogAccess.cc             | 41 +++++++++++++++++++++++++++++
 proxy/logging/LogAccess.h              |  2 ++
 proxy/logging/LogAccessHttp.cc         |  9 +++++++
 proxy/logging/LogAccessHttp.h          |  1 +
 6 files changed, 77 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/trafficserver/blob/f914a62f/doc/admin/event-logging-formats.en.rst
----------------------------------------------------------------------
diff --git a/doc/admin/event-logging-formats.en.rst b/doc/admin/event-logging-formats.en.rst
index 67d484d..751ec35 100644
--- a/doc/admin/event-logging-formats.en.rst
+++ b/doc/admin/event-logging-formats.en.rst
@@ -75,6 +75,10 @@ The following list describes Traffic Server custom logging fields.
 ``chp``
     The port number of the client's host machine.
 
+``cps``
+    Client Protocol Stack, the output would be the conjunction of
+    protocol names in the stack spliced with '+', such as "TLS+SPDY".
+
 ``cqbl``
     The client request transfer length; the body length in the client
     request to Traffic Server (in bytes).

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/f914a62f/proxy/logging/Log.cc
----------------------------------------------------------------------
diff --git a/proxy/logging/Log.cc b/proxy/logging/Log.cc
index 462d955..8a7d0d5 100644
--- a/proxy/logging/Log.cc
+++ b/proxy/logging/Log.cc
@@ -365,6 +365,26 @@ Log::init_fields()
   global_field_list.add (field, false);
   ink_hash_table_insert (field_symbol_hash, "caun", field);
 
+  Ptr<LogFieldAliasTable> proto_type_map = make_ptr(NEW(new LogFieldAliasTable));
+  proto_type_map->init(4,
+                       // Transport protocols
+                       TS_PROTO_UDP, "UDP",
+                       TS_PROTO_TCP, "TCP",
+                       TS_PROTO_TLS, "TLS",
+                       // Application protocols
+                       TS_PROTO_HTTP, "HTTP",
+                       TS_PROTO_SPDY, "SPDY",
+                       TS_PROTO_RTMP, "RTMP",
+                       TS_PROTO_WBSK, "WBSK");
+
+  field = NEW(new LogField("client_protocol_stack", "cps",
+                           LogField::sINT,
+                           &LogAccess::marshal_client_protocol_stack,
+                           &LogAccess::unmarshal_client_protocol_stack,
+                           (Ptr<LogFieldAliasMap>) proto_type_map));
+  global_field_list.add(field, false);
+  ink_hash_table_insert(field_symbol_hash, "cps", field);
+
   field = NEW(new LogField("client_req_timestamp_sec", "cqts",
                            LogField::sINT,
                            &LogAccess::marshal_client_req_timestamp_sec,

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/f914a62f/proxy/logging/LogAccess.cc
----------------------------------------------------------------------
diff --git a/proxy/logging/LogAccess.cc b/proxy/logging/LogAccess.cc
index 568c759..20121f7 100644
--- a/proxy/logging/LogAccess.cc
+++ b/proxy/logging/LogAccess.cc
@@ -75,6 +75,12 @@ LogAccess::init()
   -------------------------------------------------------------------------*/
 
 int
+LogAccess::marshal_client_protocol_stack(char *buf)
+{
+  DEFAULT_INT_FIELD;
+}
+
+int
 LogAccess::marshal_client_host_ip(char *buf)
 {
   DEFAULT_IP_FIELD;
@@ -1306,6 +1312,41 @@ LogAccess::unmarshal_cache_write_code(char **buf, char *dest, int len, Ptr<LogFi
 }
 
 int
+LogAccess::unmarshal_client_protocol_stack(char **buf, char *dest, int len, Ptr<LogFieldAliasMap> map)
+{
+  ink_assert(buf != NULL);
+  ink_assert(*buf != NULL);
+  ink_assert(dest != NULL);
+
+  char *p;
+  size_t nr_chars = 0;
+  int i, ret, nr_bits, left_len;
+  TSClientProtoStack proto_stack = (TSClientProtoStack)unmarshal_int(buf);
+
+  p = dest;
+  left_len = len;
+  nr_bits = 8 * sizeof(TSClientProtoStack);
+
+  for (i = 0; i < nr_bits && left_len; i++) {
+    if ((proto_stack >> i) & 0x1) {
+      if (p != dest) {
+        *p++ = '+';
+        left_len--;
+      }
+      ret = map->asString(i, p, left_len, &nr_chars);
+      if (ret == LogFieldAliasMap::ALL_OK) {
+        p += nr_chars;
+        left_len -= nr_chars;
+      } else if (ret == LogFieldAliasMap::BUFFER_TOO_SMALL) {
+        break;
+      }
+    }
+  }
+
+  return (len - left_len);
+}
+
+int
 LogAccess::unmarshal_record(char **buf, char *dest, int len)
 {
   ink_assert(buf != NULL);

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/f914a62f/proxy/logging/LogAccess.h
----------------------------------------------------------------------
diff --git a/proxy/logging/LogAccess.h b/proxy/logging/LogAccess.h
index a152de6..35291aa 100644
--- a/proxy/logging/LogAccess.h
+++ b/proxy/logging/LogAccess.h
@@ -172,6 +172,7 @@ public:
   inkcoreapi virtual int marshal_client_auth_user_name(char *); // STR
   int marshal_client_req_timestamp_sec(char *); // INT
 
+  inkcoreapi virtual int marshal_client_protocol_stack(char *); // INT
   inkcoreapi virtual int marshal_client_req_text(char *);       // STR
   inkcoreapi virtual int marshal_client_req_http_method(char *);        // STR
   inkcoreapi virtual int marshal_client_req_url(char *);        // STR
@@ -290,6 +291,7 @@ public:
   static int unmarshal_cache_code(char **buf, char *dest, int len, Ptr<LogFieldAliasMap> map);
   static int unmarshal_entry_type(char **buf, char *dest, int len, Ptr<LogFieldAliasMap> map);
   static int unmarshal_cache_write_code(char **buf, char *dest, int len, Ptr<LogFieldAliasMap> map);
+  static int unmarshal_client_protocol_stack(char **buf, char *dest, int len, Ptr<LogFieldAliasMap> map);
 
   static int unmarshal_with_map(int64_t code, char *dest, int len, Ptr<LogFieldAliasMap> map, const char *msg = 0);
 

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/f914a62f/proxy/logging/LogAccessHttp.cc
----------------------------------------------------------------------
diff --git a/proxy/logging/LogAccessHttp.cc b/proxy/logging/LogAccessHttp.cc
index 851458c..5ad308b 100644
--- a/proxy/logging/LogAccessHttp.cc
+++ b/proxy/logging/LogAccessHttp.cc
@@ -139,6 +139,15 @@ LogAccessHttp::init()
   -------------------------------------------------------------------------*/
 
 int
+LogAccessHttp::marshal_client_protocol_stack(char *buf)
+{
+  if (buf) {
+    marshal_int(buf, m_http_sm->proto_stack);
+  }
+  return INK_MIN_ALIGN;
+}
+
+int
 LogAccessHttp::marshal_client_host_ip(char *buf)
 {
   return marshal_ip(buf, &m_http_sm->t_state.client_info.addr.sa);

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/f914a62f/proxy/logging/LogAccessHttp.h
----------------------------------------------------------------------
diff --git a/proxy/logging/LogAccessHttp.h b/proxy/logging/LogAccessHttp.h
index 8ca3e8b..9235441 100644
--- a/proxy/logging/LogAccessHttp.h
+++ b/proxy/logging/LogAccessHttp.h
@@ -58,6 +58,7 @@ public:
   virtual int marshal_client_host_ip(char *);   // STR
   virtual int marshal_client_host_port(char *); // INT
   virtual int marshal_client_auth_user_name(char *);    // STR
+  virtual int marshal_client_protocol_stack(char *);    // INT
   virtual int marshal_client_req_text(char *);  // STR
   virtual int marshal_client_req_http_method(char *);   // INT
   virtual int marshal_client_req_url(char *);   // STR

[2/5] git commit: TS-1062: Make TSFetchUrl handle chunked encoding automatically

Posted by yu...@apache.org.

TS-1062: Make TSFetchUrl handle chunked encoding automatically

Signed-off-by: Yunkai Zhang <qi...@taobao.com>


Project: http://git-wip-us.apache.org/repos/asf/trafficserver/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafficserver/commit/6a5f55ab
Tree: http://git-wip-us.apache.org/repos/asf/trafficserver/tree/6a5f55ab
Diff: http://git-wip-us.apache.org/repos/asf/trafficserver/diff/6a5f55ab

Branch: refs/heads/master
Commit: 6a5f55abd2088e02e615639b182b757fbdc0590e
Parents: 8a0bee4
Author: Yunkai Zhang <qi...@taobao.com>
Authored: Mon Mar 3 21:44:18 2014 +0800
Committer: Yunkai Zhang <qi...@taobao.com>
Committed: Wed Mar 12 11:02:15 2014 +0800

----------------------------------------------------------------------
 proxy/FetchSM.cc | 92 ++++++++++++++++++++++++++++++++++-----------------
 proxy/FetchSM.h  |  2 +-
 2 files changed, 63 insertions(+), 31 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/trafficserver/blob/6a5f55ab/proxy/FetchSM.cc
----------------------------------------------------------------------
diff --git a/proxy/FetchSM.cc b/proxy/FetchSM.cc
index 77d3197..b869837 100644
--- a/proxy/FetchSM.cc
+++ b/proxy/FetchSM.cc
@@ -289,43 +289,81 @@ out:
 void
 FetchSM::get_info_from_buffer(IOBufferReader *the_reader)
 {
-  char *info;
-//  char *info_start;
-
+  char *buf, *info;
   int64_t read_avail, read_done;
   IOBufferBlock *blk;
-  char *buf;
+  IOBufferReader *reader = the_reader;
 
-  if (!the_reader)
+  if (!reader) {
+    client_bytes = 0;
     return ;
+  }
 
-  read_avail = the_reader->read_avail();
+  read_avail = reader->read_avail();
   Debug(DEBUG_TAG, "[%s] total avail %" PRId64 , __FUNCTION__, read_avail);
-  //size_t hdr_size = _headers.size();
-  //info = (char *)ats_malloc(sizeof(char) * (read_avail+1) + hdr_size);
+  if (!read_avail) {
+    client_bytes = 0;
+    return;
+  }
+
   info = (char *)ats_malloc(sizeof(char) * (read_avail+1));
   client_response = info;
 
-  //ink_strlcpy(info, _headers.data(), sizeof(char) * (read_avail+1));
-  //info += hdr_size;
+  if (!check_chunked()) {
+    /* Read the data out of the reader */
+    while (read_avail > 0) {
+      if (reader->block != NULL)
+        reader->skip_empty_blocks();
+      blk = reader->block;
+
+      // This is the equivalent of TSIOBufferBlockReadStart()
+      buf = blk->start() + reader->start_offset;
+      read_done = blk->read_avail() - reader->start_offset;
+
+      if (read_done > 0) {
+        memcpy(info, buf, read_done);
+        reader->consume(read_done);
+        read_avail -= read_done;
+        info += read_done;
+        client_bytes += read_done;
+      }
+    }
+    client_response[client_bytes] = '\0';
+    return;
+  }
 
-  /* Read the data out of the reader */
-  while (read_avail > 0) {
-    if (the_reader->block != NULL)
-      the_reader->skip_empty_blocks();
-    blk = the_reader->block;
+  reader = chunked_handler.dechunked_reader;
+  do {
+    if (chunked_handler.state == ChunkedHandler::CHUNK_FLOW_CONTROL) {
+      chunked_handler.state = ChunkedHandler::CHUNK_READ_SIZE_START;
+    }
 
-    // This is the equivalent of TSIOBufferBlockReadStart()
-    buf = blk->start() + the_reader->start_offset;
-    read_done = blk->read_avail() - the_reader->start_offset;
+    if (!dechunk_body())
+      break;
 
-    if (read_done > 0) {
-      memcpy(info, buf, read_done);
-      the_reader->consume(read_done);
-      read_avail -= read_done;
-      info += read_done;
+    /* Read the data out of the reader */
+    read_avail = reader->read_avail();
+    while (read_avail > 0) {
+      if (reader->block != NULL)
+        reader->skip_empty_blocks();
+      blk = reader->block;
+
+      // This is the equivalent of TSIOBufferBlockReadStart()
+      buf = blk->start() + reader->start_offset;
+      read_done = blk->read_avail() - reader->start_offset;
+
+      if (read_done > 0) {
+        memcpy(info, buf, read_done);
+        reader->consume(read_done);
+        read_avail -= read_done;
+        info += read_done;
+        client_bytes += read_done;
+      }
     }
-  }
+  } while (chunked_handler.state == ChunkedHandler::CHUNK_FLOW_CONTROL);
+
+  client_response[client_bytes] = '\0';
+  return;
 }
 
 void
@@ -360,13 +398,7 @@ FetchSM::process_fetch_read(int event)
     if (fetch_flags & TS_FETCH_FLAGS_STREAM)
       return InvokePluginExt();
     if(callback_options == AFTER_HEADER || callback_options == AFTER_BODY) {
-      bytes = resp_reader->read_avail();
       get_info_from_buffer(resp_reader);
-      Debug(DEBUG_TAG, "[%s] number of bytes %"PRId64"", __FUNCTION__, bytes);
-      if(client_response!=NULL)
-        client_response[bytes] = '\0';
-      Debug(DEBUG_TAG, "[%s] Completed data fetch of size %"PRId64", notifying caller", __FUNCTION__, bytes);
-      client_bytes = bytes;
       InvokePlugin( callback_events.success_event_id, (void *) this);
     }
     Debug(DEBUG_TAG, "[%s] received EOS", __FUNCTION__);

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/6a5f55ab/proxy/FetchSM.h
----------------------------------------------------------------------
diff --git a/proxy/FetchSM.h b/proxy/FetchSM.h
index 4a07a72..0de5d96 100644
--- a/proxy/FetchSM.h
+++ b/proxy/FetchSM.h
@@ -76,7 +76,7 @@ public:
     callback_events = events;
     callback_options = options;
     _addr.assign(addr);
-    fetch_flags = TS_FETCH_FLAGS_NONE;
+    fetch_flags = TS_FETCH_FLAGS_DECHUNK;
     writeRequest(headers,length);
     mutex = new_ProxyMutex();

[3/5] git commit: TS-1062: Extends and optimizes FetchSM

Posted by yu...@apache.org.

TS-1062: Extends and optimizes FetchSM

* Optimize FetchSM to support stream IO.
* Reduce memory copy in FetchSM.
* Expose some plugin APIs in 'ts/experimental.h'.

This patch will borrow some ideas from @Quehan's fetcher
library, which has not been open-sourced yet.

Signed-off-by: Yunkai Zhang <qi...@taobao.com>


Project: http://git-wip-us.apache.org/repos/asf/trafficserver/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafficserver/commit/45f65532
Tree: http://git-wip-us.apache.org/repos/asf/trafficserver/tree/45f65532
Diff: http://git-wip-us.apache.org/repos/asf/trafficserver/diff/45f65532

Branch: refs/heads/master
Commit: 45f6553290a62aa9e77d42e32e7809b89c41ace2
Parents: 558345f
Author: Yunkai Zhang <qi...@taobao.com>
Authored: Mon Mar 3 18:34:54 2014 +0800
Committer: Yunkai Zhang <qi...@taobao.com>
Committed: Wed Mar 12 11:02:15 2014 +0800

----------------------------------------------------------------------
 proxy/FetchSM.cc            | 434 +++++++++++++++++++++++++++++++++++----
 proxy/FetchSM.h             |  80 +++++++-
 proxy/InkAPI.cc             |  99 +++++++++
 proxy/api/ts/experimental.h | 119 +++++++++++
 4 files changed, 684 insertions(+), 48 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/trafficserver/blob/45f65532/proxy/FetchSM.cc
----------------------------------------------------------------------
diff --git a/proxy/FetchSM.cc b/proxy/FetchSM.cc
index 89e2f68..16d4f4c 100644
--- a/proxy/FetchSM.cc
+++ b/proxy/FetchSM.cc
@@ -27,20 +27,33 @@
 #include "HTTP.h"
 #include "PluginVC.h"
 
+static const char *http_method[] = {
+  "NONE",
+  "GET",
+  "POST",
+  "CONNECT",
+  "DELETE",
+  "HEAD",
+  "PURGE",
+  "PUT",
+  "LAST",
+};
+
 #define DEBUG_TAG "FetchSM"
+#define FETCH_LOCK_RETRY_TIME HRTIME_MSECONDS(10)
 
 ClassAllocator < FetchSM > FetchSMAllocator("FetchSMAllocator");
 void
 FetchSM::cleanUp()
 {
   Debug(DEBUG_TAG, "[%s] calling cleanup", __FUNCTION__);
-  free_MIOBuffer(response_buffer);
   free_MIOBuffer(req_buffer);
   free_MIOBuffer(resp_buffer);
   mutex.clear();
   http_parser_clear(&http_parser);
   client_response_hdr.destroy();
   ats_free(client_response);
+  cont_mutex.clear();
 
   PluginVC *vc = (PluginVC *) http_vc;
 
@@ -57,7 +70,7 @@ FetchSM::httpConnect()
   PluginVC *vc = (PluginVC *) http_vc;
 
   read_vio = vc->do_io_read(this, INT64_MAX, resp_buffer);
-  write_vio = vc->do_io_write(this, getReqLen(), req_reader);
+  write_vio = vc->do_io_write(this, getReqLen() + req_content_length, req_reader);
 }
 
 char* FetchSM::resp_get(int *length) {
@@ -65,7 +78,8 @@ char* FetchSM::resp_get(int *length) {
   return client_response;
 }
 
-int FetchSM::InvokePlugin(int event, void *data)
+int
+FetchSM::InvokePlugin(int event, void *data)
 {
   EThread *mythread = this_ethread();
 
@@ -77,6 +91,174 @@ int FetchSM::InvokePlugin(int event, void *data)
 
   return ret;
 }
+
+bool
+FetchSM::has_body()
+{
+  int status_code;
+  HTTPHdr *hdr;
+
+  if (!header_done)
+    return false;
+
+  //
+  // The following code comply with HTTP/1.1:
+  // http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.4
+  //
+
+  if (req_method == TS_FETCH_METHOD_HEAD)
+    return false;
+
+  hdr = &client_response_hdr;
+
+  status_code = hdr->status_get();
+  if (status_code < 200 || status_code == 204 || status_code == 304)
+    return false;
+
+  if (check_chunked())
+    return true;
+
+  resp_content_length = hdr->value_get_int64(MIME_FIELD_CONTENT_LENGTH, MIME_LEN_CONTENT_LENGTH);
+  if (!resp_content_length)
+    return false;
+
+  return true;
+}
+
+bool
+FetchSM::check_body_done()
+{
+  if (!check_chunked()) {
+    if (resp_content_length == resp_recived_body_len + resp_reader->read_avail())
+      return true;
+
+    return false;
+  }
+
+  //
+  // TODO: check whether the chunked body is done
+  //
+  return true;
+}
+
+bool
+FetchSM::check_chunked()
+{
+  int ret;
+  StrList slist;
+  HTTPHdr *hdr = &client_response_hdr;
+
+  if (resp_is_chunked >= 0)
+    return resp_is_chunked;
+
+  ink_release_assert(header_done);
+
+  resp_is_chunked = 0;
+  ret = hdr->value_get_comma_list(MIME_FIELD_TRANSFER_ENCODING,
+                                  MIME_LEN_TRANSFER_ENCODING, &slist);
+  if (ret) {
+    for (Str *f = slist.head; f != NULL; f = f->next) {
+      if (f->len == 0)
+        continue;
+
+      size_t len = sizeof("chunked") - 1;
+      len = len > f->len ? f->len : len;
+      if (!strncasecmp(f->str, "chunked", len)) {
+        resp_is_chunked = 1;
+        return true;
+      }
+    }
+  }
+
+  return resp_is_chunked;
+}
+
+int
+FetchSM::dechunk_body()
+{
+  ink_assert(resp_is_chunked > 0);
+  //
+  // TODO: dechunk the body content.
+  // return:
+  //  - 0: need to read more data.
+  //  - TS_FETCH_EVENT_EXT_BODY_READY.
+  //  - TS_FETCH_EVENT_EXT_BODY_DONE.
+  //
+  return TS_FETCH_EVENT_EXT_BODY_DONE;
+}
+
+void
+FetchSM::InvokePluginExt(int error_event)
+{
+  int event;
+  EThread *mythread = this_ethread();
+
+  //
+  // Increasing *recursion* to prevent
+  // FetchSM being deleted by callback.
+  //
+  recursion++;
+
+  if (fetch_flags & TS_FETCH_FLAGS_NEWLOCK) {
+    MUTEX_TAKE_LOCK(cont_mutex, mythread);
+  }
+
+  if (!contp)
+    goto out;
+
+  if (error_event) {
+    contp->handleEvent(error_event, this);
+    goto out;
+  }
+
+  if (!has_sent_header) {
+    contp->handleEvent(TS_FETCH_EVENT_EXT_HEAD_DONE, this);
+    has_sent_header = true;
+  }
+
+  if (!has_body()) {
+    contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_DONE, this);
+    goto out;
+  }
+
+  if (!resp_reader->read_avail())
+    goto out;
+
+  Debug(DEBUG_TAG, "[%s] chunked:%d, content_len:%ld, recived_len:%ld, avail:%ld\n",
+        __FUNCTION__, resp_is_chunked, resp_content_length, resp_recived_body_len, resp_reader->read_avail());
+
+  if (!check_chunked()) {
+    if (!check_body_done())
+      contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_READY, this);
+    else
+      contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_DONE, this);
+  } else if (fetch_flags & TS_FETCH_FLAGS_DECHUNK){
+    event = dechunk_body();
+
+    if (!event) {
+      read_vio->reenable();
+      goto out;
+    }
+
+    contp->handleEvent(event, this);
+  } else if (check_body_done()){
+    contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_DONE, this);
+  } else {
+    contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_READY, this);
+  }
+
+out:
+  if (fetch_flags & TS_FETCH_FLAGS_NEWLOCK) {
+    MUTEX_UNTAKE_LOCK(cont_mutex, mythread);
+  }
+  recursion--;
+
+  if (!contp && !recursion)
+    cleanUp();
+
+  return;
+}
+
 void
 FetchSM::get_info_from_buffer(IOBufferReader *the_reader)
 {
@@ -125,57 +307,51 @@ FetchSM::process_fetch_read(int event)
   Debug(DEBUG_TAG, "[%s] I am here read", __FUNCTION__);
   int64_t bytes;
   int bytes_used;
-  int64_t total_bytes_copied = 0;
 
   switch (event) {
   case TS_EVENT_VCONN_READ_READY:
     bytes = resp_reader->read_avail();
-    Debug(DEBUG_TAG, "[%s] number of bytes in read ready %" PRId64"", __FUNCTION__, bytes);
-    while (total_bytes_copied < bytes) {
-       int64_t actual_bytes_copied;
-       actual_bytes_copied = response_buffer->write(resp_reader, bytes, 0);
-       if (actual_bytes_copied <= 0) {
-           break;
-       }
-       total_bytes_copied += actual_bytes_copied;
-    }
-    resp_reader->consume(total_bytes_copied);
-    if (header_done == 0 && callback_options == AFTER_HEADER) {
-      if (client_response_hdr.parse_resp(&http_parser, response_reader, &bytes_used, 0) == PARSE_DONE) {
-        //InvokePlugin( TS_EVENT_INTERNAL_60201, (void *) &client_response_hdr);
-        InvokePlugin( callback_events.success_event_id, (void *) &client_response_hdr);
+    Debug(DEBUG_TAG, "[%s] number of bytes in read ready %"PRId64"", __FUNCTION__, bytes);
+    if (header_done == 0 && ((fetch_flags & TS_FETCH_FLAGS_STREAM) || callback_options == AFTER_HEADER)) {
+      if (client_response_hdr.parse_resp(&http_parser, resp_reader, &bytes_used, 0) == PARSE_DONE) {
         header_done = 1;
+        if (fetch_flags & TS_FETCH_FLAGS_STREAM)
+          return InvokePluginExt();
+        else
+          InvokePlugin( callback_events.success_event_id, (void *) &client_response_hdr);
       }
+    } else {
+      if (fetch_flags & TS_FETCH_FLAGS_STREAM)
+        return InvokePluginExt();
+      else
+        InvokePlugin(TS_FETCH_EVENT_EXT_BODY_READY, this);
     }
     read_vio->reenable();
     break;
   case TS_EVENT_VCONN_READ_COMPLETE:
   case TS_EVENT_VCONN_EOS:
+    if (fetch_flags & TS_FETCH_FLAGS_STREAM)
+      return InvokePluginExt();
     if(callback_options == AFTER_HEADER || callback_options == AFTER_BODY) {
-    bytes = response_reader->read_avail();
-
-    get_info_from_buffer(response_reader);
-    Debug(DEBUG_TAG, "[%s] number of bytes %" PRId64"", __FUNCTION__, bytes);
-    if(client_response!=NULL)
-      client_response[bytes] = '\0';
-      //client_response[bytes + _headers.size()] = '\0';
-    Debug(DEBUG_TAG, "[%s] Completed data fetch of size %" PRId64", notifying caller", __FUNCTION__, bytes);
-    //InvokePlugin( TS_EVENT_INTERNAL_60200, (void *) client_response);
-   client_bytes = bytes;
-    //InvokePlugin( TS_EVENT_INTERNAL_60200, (void *) this);
+      bytes = resp_reader->read_avail();
+      get_info_from_buffer(resp_reader);
+      Debug(DEBUG_TAG, "[%s] number of bytes %"PRId64"", __FUNCTION__, bytes);
+      if(client_response!=NULL)
+        client_response[bytes] = '\0';
+      Debug(DEBUG_TAG, "[%s] Completed data fetch of size %"PRId64", notifying caller", __FUNCTION__, bytes);
+      client_bytes = bytes;
       InvokePlugin( callback_events.success_event_id, (void *) this);
     }
-
     Debug(DEBUG_TAG, "[%s] received EOS", __FUNCTION__);
     cleanUp();
     break;
   case TS_EVENT_ERROR:
   default:
-    //InvokePlugin(TS_EVENT_ERROR, NULL);
-      InvokePlugin( callback_events.failure_event_id, NULL);
+    if (fetch_flags & TS_FETCH_FLAGS_STREAM)
+      return InvokePluginExt(event);
+    InvokePlugin( callback_events.failure_event_id, NULL);
     cleanUp();
     break;
-
   }
 }
 
@@ -185,18 +361,19 @@ FetchSM::process_fetch_write(int event)
   Debug(DEBUG_TAG, "[%s] calling process write", __FUNCTION__);
   switch (event) {
   case TS_EVENT_VCONN_WRITE_COMPLETE:
-    //INKVConnShutdown(http_vc, 0, 1) ; why does not this work???
     req_finished = true;
     break;
   case TS_EVENT_VCONN_WRITE_READY:
     // data is processed in chunks of 32k; if there is more than 32k
     // of input data, we have to continue reenabling until all data is
     // read (we have already written all the data to the buffer)
-    ((PluginVC *) http_vc)->reenable(write_vio);
+    if (req_reader->read_avail() > 0)
+      ((PluginVC *) http_vc)->reenable(write_vio);
     break;
   case TS_EVENT_ERROR:
-    //InvokePlugin( TS_EVENT_ERROR, NULL);
-      InvokePlugin( callback_events.failure_event_id, NULL);
+    if (fetch_flags & TS_FETCH_FLAGS_STREAM)
+      return InvokePluginExt(event);
+    InvokePlugin( callback_events.failure_event_id, NULL);
     cleanUp();
   default:
     break;
@@ -213,8 +390,189 @@ FetchSM::fetch_handler(int event, void *edata)
   } else if (edata == write_vio) {
     process_fetch_write(event);
   } else {
-      InvokePlugin( callback_events.failure_event_id, NULL);
+    if (fetch_flags & TS_FETCH_FLAGS_STREAM) {
+      InvokePluginExt(event);
+      return 1;
+    }
+    InvokePlugin( callback_events.failure_event_id, NULL);
     cleanUp();
   }
   return 1;
 }
+
+void
+FetchSM::ext_init(Continuation *cont, TSFetchMethod method,
+                  const char *url, const char *version,
+                  const sockaddr *client_addr, int flags)
+{
+  init_comm();
+
+  if (flags & TS_FETCH_FLAGS_NEWLOCK) {
+    mutex = new_ProxyMutex();
+    cont_mutex = cont->mutex;
+  } else {
+    mutex = cont->mutex;
+  }
+
+  contp = cont;
+  _addr.assign(client_addr);
+
+  //
+  // Enable stream IO automatically.
+  //
+  fetch_flags = (TS_FETCH_FLAGS_STREAM | flags);
+
+  //
+  // These options are not used when enable
+  // stream IO.
+  //
+  memset(&callback_options, 0, sizeof(callback_options));
+  memset(&callback_events, 0, sizeof(callback_events));
+
+  req_method = method;
+  req_buffer->write(http_method[method], strlen(http_method[method]));
+  req_buffer->write(" ", 1);
+  req_buffer->write(url, strlen(url));
+  req_buffer->write(" ", 1);
+  req_buffer->write(version, strlen(version));
+  req_buffer->write("\r\n", 2);
+}
+
+void
+FetchSM::ext_add_header(const char *name, int name_len,
+                        const char *value, int value_len)
+{
+  if (TS_MIME_LEN_CONTENT_LENGTH == name_len &&
+      !strncasecmp(TS_MIME_FIELD_CONTENT_LENGTH, name, name_len)) {
+    req_content_length = atoll(value);
+  }
+
+  req_buffer->write(name, name_len);
+  req_buffer->write(": ", 2);
+  req_buffer->write(value, value_len);
+  req_buffer->write("\r\n", 2);
+}
+
+void
+FetchSM::ext_lanuch()
+{
+  req_buffer->write("\r\n", 2);
+  httpConnect();
+}
+
+void
+FetchSM::ext_write_data(const void *data, size_t len)
+{
+  if (header_done && (fetch_flags & TS_FETCH_FLAGS_NEWLOCK)) {
+    MUTEX_TAKE_LOCK(mutex, this_ethread());
+  }
+
+  req_buffer->write(data, len);
+
+  //
+  // Before header_done, FetchSM may not
+  // be initialized.
+  //
+  if (header_done)
+    write_vio->reenable();
+
+  if (header_done && (fetch_flags & TS_FETCH_FLAGS_NEWLOCK)) {
+    MUTEX_UNTAKE_LOCK(mutex, this_ethread());
+  }
+}
+
+ssize_t
+FetchSM::ext_read_data(char *buf, size_t len)
+{
+  const char *start;
+  TSIOBufferReader reader;
+  TSIOBufferBlock blk, next_blk;
+  int64_t already, blk_len, need, wavail;
+
+  if (fetch_flags & TS_FETCH_FLAGS_NEWLOCK) {
+    MUTEX_TRY_LOCK(lock, mutex, this_ethread());
+    if (!lock)
+      return 0;
+  }
+
+  if (!header_done)
+    return 0;
+
+  if (check_chunked())
+    reader = NULL; // TODO: asign dechunking reader
+  else
+    reader = (TSIOBufferReader)resp_reader;
+
+  already = 0;
+  blk = TSIOBufferReaderStart(reader);
+
+  while (blk) {
+
+    wavail = len - already;
+
+    next_blk = TSIOBufferBlockNext(blk);
+    start = TSIOBufferBlockReadStart(blk, reader, &blk_len);
+
+    need = blk_len > wavail ? wavail : blk_len;
+
+    memcpy(&buf[already], start, need);
+    already += need;
+
+    if (already >= (int64_t)len)
+      break;
+
+    blk = next_blk;
+  }
+
+  resp_recived_body_len += already;
+  resp_reader->consume(already);
+
+  read_vio->reenable();
+  return already;
+}
+
+void
+FetchSM::ext_destroy()
+{
+  contp = NULL;
+
+  if (recursion)
+    return;
+
+  if (fetch_flags & TS_FETCH_FLAGS_NEWLOCK) {
+    MUTEX_TRY_LOCK(lock, mutex, this_ethread());
+    if (!lock) {
+      eventProcessor.schedule_in(this, FETCH_LOCK_RETRY_TIME);
+      return;
+    }
+  }
+
+  cleanUp();
+}
+
+void
+FetchSM::ext_set_user_data(void *data)
+{
+  user_data = data;
+}
+
+void*
+FetchSM::ext_get_user_data()
+{
+  return user_data;
+}
+
+TSMBuffer
+FetchSM::resp_hdr_bufp()
+{
+  HdrHeapSDKHandle *heap;
+  heap = (HdrHeapSDKHandle *)&client_response_hdr;
+
+  return (TSMBuffer)heap;
+}
+
+TSMLoc
+FetchSM::resp_hdr_mloc()
+{
+  return (TSMLoc)client_response_hdr.m_http;
+}

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/45f65532/proxy/FetchSM.h
----------------------------------------------------------------------
diff --git a/proxy/FetchSM.h b/proxy/FetchSM.h
index 3057fc4..1e312ef 100644
--- a/proxy/FetchSM.h
+++ b/proxy/FetchSM.h
@@ -40,30 +40,56 @@ public:
   FetchSM()
   { }
 
-  void init(Continuation* cont, TSFetchWakeUpOptions options, TSFetchEvent events, const char* headers, int length, sockaddr const * addr)
+  void init_comm()
   {
-    //_headers.assign(headers);
-    Debug("FetchSM", "[%s] FetchSM initialized for request with headers\n--\n%.*s\n--", __FUNCTION__, length, headers);
+    recursion = 0;
     req_finished = 0;
     resp_finished = 0;
     header_done = 0;
+    user_data = NULL;
+    has_sent_header = false;
+    req_method = TS_FETCH_METHOD_NONE;
+    req_content_length = 0;
+    resp_is_chunked = -1;
+    resp_content_length = -1;
+    resp_recived_body_len = 0;
+    cont_mutex.clear();
     req_buffer = new_MIOBuffer(HTTP_HEADER_BUFFER_SIZE_INDEX);
     req_reader = req_buffer->alloc_reader();
     resp_buffer = new_MIOBuffer(BUFFER_SIZE_INDEX_32K);
     resp_reader = resp_buffer->alloc_reader();
-    response_buffer = new_MIOBuffer(BUFFER_SIZE_INDEX_32K);
-    response_reader = response_buffer->alloc_reader();
-    contp = cont;
     http_parser_init(&http_parser);
     client_response_hdr.create(HTTP_TYPE_RESPONSE);
     client_response  = NULL;
-    mutex = new_ProxyMutex();
+    SET_HANDLER(&FetchSM::fetch_handler);
+  }
+
+  void init(Continuation* cont, TSFetchWakeUpOptions options,
+            TSFetchEvent events, const char* headers, int length,
+            sockaddr const *addr)
+  {
+    Debug("FetchSM", "[%s] FetchSM initialized for request with headers\n--\n%.*s\n--",
+          __FUNCTION__, length, headers);
+    init_comm();
+    contp = cont;
     callback_events = events;
     callback_options = options;
     _addr.assign(addr);
+    fetch_flags = TS_FETCH_FLAGS_NONE;
     writeRequest(headers,length);
-    SET_HANDLER(&FetchSM::fetch_handler);
+    mutex = new_ProxyMutex();
+
+    //
+    // We had dropped response_buffer/respone_reader to avoid unnecessary
+    // memory copying. But for the original TSFetchURL() API, PluginVC may
+    // stop adding data to resp_buffer when the pending data in resp_buffer
+    // reach its water_mark.
+    //
+    // So we should set the water_mark of resp_buffer with a large value,
+    // INT64_MAX would be reasonable.
+    resp_buffer->water_mark = INT64_MAX;
   }
+
   int fetch_handler(int event, void *data);
   void process_fetch_read(int event);
   void process_fetch_write(int event);
@@ -72,8 +98,29 @@ public:
   void get_info_from_buffer(IOBufferReader *reader);
   char* resp_get(int* length);
 
+  TSMBuffer resp_hdr_bufp();
+  TSMLoc resp_hdr_mloc();
+
+  //
+  // Extended APIs for FetchSM
+  //
+  // *flags* can be bitwise OR of several TSFetchFlags
+  //
+  void ext_init(Continuation *cont, TSFetchMethod method,
+                const char *url, const char *version,
+                const sockaddr *client_addr, int flags);
+  void ext_add_header(const char *name, int name_len,
+                      const char *value, int value_len);
+  void ext_lanuch();
+  void ext_destroy();
+  ssize_t ext_read_data(char *buf, size_t len);
+  void ext_write_data(const void *data, size_t len);
+  void ext_set_user_data(void *data);
+  void* ext_get_user_data();
+
 private:
   int InvokePlugin(int event, void*data);
+  void InvokePluginExt(int error_event = 0);
 
   void writeRequest(const char *headers,int length)
   {
@@ -85,11 +132,15 @@ private:
 
   int64_t getReqLen() const { return req_reader->read_avail(); }
 
+  bool has_body();
+  bool check_body_done();
+  bool check_chunked();
+  int dechunk_body();
+
+  int recursion;
   TSVConn http_vc;
   VIO *read_vio;
   VIO *write_vio;
-  MIOBuffer *response_buffer;   // response to FetchSM call
-  IOBufferReader *response_reader;      // response to FetchSM call
   MIOBuffer *req_buffer;
   IOBufferReader *req_reader;
   char *client_response;
@@ -97,6 +148,7 @@ private:
   MIOBuffer *resp_buffer;       // response to HttpConnect Call
   IOBufferReader *resp_reader;
   Continuation *contp;
+  Ptr<ProxyMutex> cont_mutex;
   HTTPParser http_parser;
   HTTPHdr client_response_hdr;
   TSFetchEvent callback_events;
@@ -105,6 +157,14 @@ private:
   bool header_done;
   bool resp_finished;
   IpEndpoint _addr;
+  int resp_is_chunked;
+  int fetch_flags;
+  void *user_data;
+  bool has_sent_header;
+  TSFetchMethod req_method;
+  int64_t req_content_length;
+  int64_t resp_content_length;
+  int64_t resp_recived_body_len;
 };
 
 #endif

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/45f65532/proxy/InkAPI.cc
----------------------------------------------------------------------
diff --git a/proxy/InkAPI.cc b/proxy/InkAPI.cc
index 9719990..a0f1c24 100644
--- a/proxy/InkAPI.cc
+++ b/proxy/InkAPI.cc
@@ -585,6 +585,15 @@ sdk_sanity_check_continuation(TSCont cont)
 }
 
 TSReturnCode
+sdk_sanity_check_fetch_sm(TSFetchSM fetch_sm)
+{
+  if (fetch_sm == NULL)
+    return TS_ERROR;
+
+  return TS_SUCCESS;
+}
+
+TSReturnCode
 sdk_sanity_check_http_ssn(TSHttpSsn ssnp)
 {
   if (ssnp == NULL)
@@ -7216,6 +7225,96 @@ TSFetchUrl(const char* headers, int request_len, sockaddr const* ip , TSCont con
   fetch_sm->httpConnect();
 }
 
+TSFetchSM
+TSFetchCreate(TSCont contp, TSFetchMethod method,
+              const char *url, const char *version,
+              struct sockaddr const* client_addr, int flags)
+{
+  sdk_assert(sdk_sanity_check_continuation(contp) == TS_SUCCESS);
+  sdk_assert(ats_is_ip4(client_addr));
+
+  FetchSM *fetch_sm = FetchSMAllocator.alloc();
+
+  fetch_sm->ext_init((Continuation*)contp, method, url, version,
+                     client_addr, flags);
+
+  return (TSFetchSM)fetch_sm;
+}
+
+void
+TSFetchHeaderAdd(TSFetchSM fetch_sm,
+                 const char *name, int name_len,
+                 const char *value, int value_len)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  ((FetchSM*)fetch_sm)->ext_add_header(name, name_len, value, value_len);
+}
+
+void
+TSFetchWriteData(TSFetchSM fetch_sm, const void *data, size_t len)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  ((FetchSM*)fetch_sm)->ext_write_data(data, len);
+}
+
+ssize_t
+TSFetchReadData(TSFetchSM fetch_sm, void *buf, size_t len)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  return ((FetchSM*)fetch_sm)->ext_read_data((char *)buf, len);
+}
+
+void
+TSFetchLaunch(TSFetchSM fetch_sm)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  ((FetchSM*)fetch_sm)->ext_lanuch();
+}
+
+void
+TSFetchDestroy(TSFetchSM fetch_sm)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  ((FetchSM*)fetch_sm)->ext_destroy();
+}
+
+void
+TSFetchUserDataSet(TSFetchSM fetch_sm, void *data)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  ((FetchSM*)fetch_sm)->ext_set_user_data(data);
+}
+
+void*
+TSFetchUserDataGet(TSFetchSM fetch_sm)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  return ((FetchSM*)fetch_sm)->ext_get_user_data();
+}
+
+TSMBuffer
+TSFetchRespHdrMBufGet(TSFetchSM fetch_sm)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  return ((FetchSM*)fetch_sm)->resp_hdr_bufp();
+}
+
+TSMLoc
+TSFetchRespHdrMLocGet(TSFetchSM fetch_sm)
+{
+  sdk_assert(sdk_sanity_check_fetch_sm(fetch_sm) == TS_SUCCESS);
+
+  return ((FetchSM*)fetch_sm)->resp_hdr_mloc();
+}
+
 TSReturnCode
 TSHttpIsInternalRequest(TSHttpTxn txnp)
 {

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/45f65532/proxy/api/ts/experimental.h
----------------------------------------------------------------------
diff --git a/proxy/api/ts/experimental.h b/proxy/api/ts/experimental.h
index d926fd8..8621158 100644
--- a/proxy/api/ts/experimental.h
+++ b/proxy/api/ts/experimental.h
@@ -37,6 +37,37 @@ extern "C"
 {
 #endif                          /* __cplusplus */
 
+  /* For Extended FetchSM APIs */
+  typedef enum {
+    TS_FETCH_METHOD_NONE,
+    TS_FETCH_METHOD_GET,
+    TS_FETCH_METHOD_POST,
+    TS_FETCH_METHOD_CONNECT,
+    TS_FETCH_METHOD_DELETE,
+    TS_FETCH_METHOD_HEAD,
+    TS_FETCH_METHOD_PURGE,
+    TS_FETCH_METHOD_PUT,
+    TS_FETCH_METHOD_LAST
+  } TSFetchMethod;
+
+  typedef enum
+  {
+    TS_FETCH_EVENT_EXT_HEAD_READY = -1,
+    TS_FETCH_EVENT_EXT_HEAD_DONE = -2,
+    TS_FETCH_EVENT_EXT_BODY_READY = -3,
+    TS_FETCH_EVENT_EXT_BODY_DONE = -4,
+  } TSFetchEventExt;
+
+  typedef enum
+  {
+    TS_FETCH_FLAGS_NONE = 0,           // do nothing
+    TS_FETCH_FLAGS_STREAM = 1 << 1,    // enable stream IO
+    TS_FETCH_FLAGS_DECHUNK = 1 << 2,   // dechunk body content
+    TS_FETCH_FLAGS_NEWLOCK = 1 << 3,   // allocate new lock for fetch sm
+  } TSFetchFlags;
+
+  typedef struct tsapi_fetchsm* TSFetchSM;
+
   /* Forward declaration of in_addr, any user of these APIs should probably 
      include net/netinet.h or whatever is appropriate on the platform. */
   struct in_addr;
@@ -600,6 +631,94 @@ extern "C"
      return value 0 indicates success */
   tsapi int TSPrefetchHookSet(int hook_no, TSPrefetchHook hook_fn);
 
+
+  /**
+   * Extended FetchSM's AIPs
+   */
+
+  /*
+   * Create FetchSM, this API will enable stream IO automatically.
+   *
+   * @param contp: continuation to be callbacked.
+   * @param method: request method.
+   * @param url: scheme://host[:port]/path.
+   * @param version: client http version, eg: "HTTP/1.1".
+   * @param client_addr: client addr sent to log.
+   * @param flags: can be bitwise OR of several TSFetchFlags.
+   *
+   * return TSFetchSM which should be destroyed by TSFetchDestroy().
+   */
+  tsapi TSFetchSM TSFetchCreate(TSCont contp, TSFetchMethod method,
+                                const char *url, const char *version,
+                                struct sockaddr const* client_addr, int flags);
+
+  /*
+   * Create FetchSM, this API will enable stream IO automatically.
+   *
+   * @param fetch_sm: returned value of TSFetchCreate().
+   * @param name: name of header.
+   * @param name_len: len of name.
+   * @param value: value of header.
+   * @param name_len: len of value.
+   *
+   * return TSFetchSM which should be destroyed by TSFetchDestroy().
+   */
+  tsapi void TSFetchHeaderAdd(TSFetchSM fetch_sm,
+                              const char *name, int name_len,
+                              const char *value, int value_len);
+
+  /*
+   * Write data to FetchSM
+   *
+   * @param fetch_sm: returned value of TSFetchCreate().
+   * @param data/len: data to be written to fetch sm.
+   */
+  tsapi void TSFetchWriteData(TSFetchSM fetch_sm, const void *data, size_t len);
+
+  /*
+   * Read up to *len* bytes from FetchSM into *buf*.
+   *
+   * @param fetch_sm: returned value of TSFetchCreate().
+   * @param buf/len: buffer to contain data from fetch sm.
+   */
+  tsapi ssize_t TSFetchReadData(TSFetchSM fetch_sm, void *buf, size_t len);
+
+  /*
+   * Lanuch FetchSM to do http request, before calling this API,
+   * you should append http request header into fetch sm through
+   * TSFetchWriteData() API
+   *
+   * @param fetch_sm: comes from returned value of TSFetchCreate().
+   */
+  tsapi void TSFetchLaunch(TSFetchSM fetch_sm);
+
+  /*
+   * Destroy FetchSM
+   *
+   * @param fetch_sm: returned value of TSFetchCreate().
+   */
+  tsapi void TSFetchDestroy(TSFetchSM fetch_sm);
+
+  /*
+   * Set user-defined data in FetchSM
+   */
+  tsapi void TSFetchUserDataSet(TSFetchSM fetch_sm, void *data);
+
+  /*
+   * Get user-defined data in FetchSM
+   */
+  tsapi void* TSFetchUserDataGet(TSFetchSM fetch_sm);
+
+  /*
+   * Get client response hdr mbuffer
+   */
+  tsapi TSMBuffer TSFetchRespHdrMBufGet(TSFetchSM fetch_sm);
+
+  /*
+   * Get client response hdr mloc
+   */
+  tsapi TSMLoc TSFetchRespHdrMLocGet(TSFetchSM fetch_sm);
+
 #ifdef __cplusplus
 }
 #endif                          /* __cplusplus */

[5/5] git commit: TS-1062: Implement dechunk supporting for extended FetchSM

Posted by yu...@apache.org.

TS-1062: Implement dechunk supporting for extended FetchSM

With this patch, we can let FetchSM to dechunk body content automatically
by passing 'TS_FETCH_FLAGS_DECHUNK' option to TSFetchCreate();

Signed-off-by: Yunkai Zhang <qi...@taobao.com>


Project: http://git-wip-us.apache.org/repos/asf/trafficserver/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafficserver/commit/8a0bee4e
Tree: http://git-wip-us.apache.org/repos/asf/trafficserver/tree/8a0bee4e
Diff: http://git-wip-us.apache.org/repos/asf/trafficserver/diff/8a0bee4e

Branch: refs/heads/master
Commit: 8a0bee4e2b4e897c8e9b089bdbc409999338bdd2
Parents: 45f6553
Author: Yunkai Zhang <qi...@taobao.com>
Authored: Mon Mar 3 18:35:37 2014 +0800
Committer: Yunkai Zhang <qi...@taobao.com>
Committed: Wed Mar 12 11:02:15 2014 +0800

----------------------------------------------------------------------
 proxy/FetchSM.cc         | 59 +++++++++++++++++++++++++++++++------------
 proxy/FetchSM.h          |  2 ++
 proxy/http/HttpTunnel.cc | 52 ++++++++++++++++++++++++++++++++------
 proxy/http/HttpTunnel.h  | 15 ++++++++---
 4 files changed, 101 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/trafficserver/blob/8a0bee4e/proxy/FetchSM.cc
----------------------------------------------------------------------
diff --git a/proxy/FetchSM.cc b/proxy/FetchSM.cc
index 16d4f4c..77d3197 100644
--- a/proxy/FetchSM.cc
+++ b/proxy/FetchSM.cc
@@ -47,6 +47,11 @@ void
 FetchSM::cleanUp()
 {
   Debug(DEBUG_TAG, "[%s] calling cleanup", __FUNCTION__);
+
+  if (resp_is_chunked > 0 && (fetch_flags & TS_FETCH_FLAGS_DECHUNK)) {
+    chunked_handler.clear();
+   }
+
   free_MIOBuffer(req_buffer);
   free_MIOBuffer(resp_buffer);
   mutex.clear();
@@ -165,6 +170,13 @@ FetchSM::check_chunked()
       len = len > f->len ? f->len : len;
       if (!strncasecmp(f->str, "chunked", len)) {
         resp_is_chunked = 1;
+        if (fetch_flags & TS_FETCH_FLAGS_DECHUNK) {
+          ChunkedHandler *ch = &chunked_handler;
+          ch->init_by_action(resp_reader, ChunkedHandler::ACTION_DECHUNK);
+          ch->dechunked_reader = ch->dechunked_buffer->alloc_reader();
+          ch->state = ChunkedHandler::CHUNK_READ_SIZE;
+          resp_reader->dealloc();
+        }
         return true;
       }
     }
@@ -178,13 +190,18 @@ FetchSM::dechunk_body()
 {
   ink_assert(resp_is_chunked > 0);
   //
-  // TODO: dechunk the body content.
-  // return:
+  // Return Value:
   //  - 0: need to read more data.
   //  - TS_FETCH_EVENT_EXT_BODY_READY.
   //  - TS_FETCH_EVENT_EXT_BODY_DONE.
   //
-  return TS_FETCH_EVENT_EXT_BODY_DONE;
+  if (chunked_handler.process_chunked_content())
+    return TS_FETCH_EVENT_EXT_BODY_DONE;
+
+  if (chunked_handler.dechunked_reader->read_avail())
+    return TS_FETCH_EVENT_EXT_BODY_READY;
+
+  return 0;
 }
 
 void
@@ -221,11 +238,16 @@ FetchSM::InvokePluginExt(int error_event)
     goto out;
   }
 
-  if (!resp_reader->read_avail())
-    goto out;
-
   Debug(DEBUG_TAG, "[%s] chunked:%d, content_len:%ld, recived_len:%ld, avail:%ld\n",
-        __FUNCTION__, resp_is_chunked, resp_content_length, resp_recived_body_len, resp_reader->read_avail());
+        __FUNCTION__, resp_is_chunked, resp_content_length, resp_recived_body_len,
+        resp_is_chunked > 0 ? chunked_handler.chunked_reader->read_avail() : resp_reader->read_avail());
+
+  if (resp_is_chunked > 0) {
+    if (!chunked_handler.chunked_reader->read_avail())
+      goto out;
+  } else if (!resp_reader->read_avail()) {
+      goto out;
+  }
 
   if (!check_chunked()) {
     if (!check_body_done())
@@ -233,14 +255,19 @@ FetchSM::InvokePluginExt(int error_event)
     else
       contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_DONE, this);
   } else if (fetch_flags & TS_FETCH_FLAGS_DECHUNK){
-    event = dechunk_body();
+    do {
+      if (chunked_handler.state == ChunkedHandler::CHUNK_FLOW_CONTROL) {
+        chunked_handler.state = ChunkedHandler::CHUNK_READ_SIZE_START;
+      }
 
-    if (!event) {
-      read_vio->reenable();
-      goto out;
-    }
+      event = dechunk_body();
+      if (!event) {
+        read_vio->reenable();
+        goto out;
+      }
 
-    contp->handleEvent(event, this);
+      contp->handleEvent(event, this);
+    } while (chunked_handler.state == ChunkedHandler::CHUNK_FLOW_CONTROL);
   } else if (check_body_done()){
     contp->handleEvent(TS_FETCH_EVENT_EXT_BODY_DONE, this);
   } else {
@@ -498,8 +525,8 @@ FetchSM::ext_read_data(char *buf, size_t len)
   if (!header_done)
     return 0;
 
-  if (check_chunked())
-    reader = NULL; // TODO: asign dechunking reader
+  if (check_chunked() && (fetch_flags & TS_FETCH_FLAGS_DECHUNK))
+    reader = (tsapi_bufferreader*)chunked_handler.dechunked_reader;
   else
     reader = (TSIOBufferReader)resp_reader;
 
@@ -525,7 +552,7 @@ FetchSM::ext_read_data(char *buf, size_t len)
   }
 
   resp_recived_body_len += already;
-  resp_reader->consume(already);
+  TSIOBufferReaderConsume(reader, already);
 
   read_vio->reenable();
   return already;

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/8a0bee4e/proxy/FetchSM.h
----------------------------------------------------------------------
diff --git a/proxy/FetchSM.h b/proxy/FetchSM.h
index 1e312ef..4a07a72 100644
--- a/proxy/FetchSM.h
+++ b/proxy/FetchSM.h
@@ -33,6 +33,7 @@
 #include "P_Net.h"
 #include "ts.h"
 #include "HttpSM.h"
+#include "HttpTunnel.h"
 
 class FetchSM: public Continuation
 {
@@ -151,6 +152,7 @@ private:
   Ptr<ProxyMutex> cont_mutex;
   HTTPParser http_parser;
   HTTPHdr client_response_hdr;
+  ChunkedHandler chunked_handler;
   TSFetchEvent callback_events;
   TSFetchWakeUpOptions callback_options;
   bool req_finished;

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/8a0bee4e/proxy/http/HttpTunnel.cc
----------------------------------------------------------------------
diff --git a/proxy/http/HttpTunnel.cc b/proxy/http/HttpTunnel.cc
index ca2f60c..a22b1c1 100644
--- a/proxy/http/HttpTunnel.cc
+++ b/proxy/http/HttpTunnel.cc
@@ -70,27 +70,63 @@ ChunkedHandler::ChunkedHandler()
 void
 ChunkedHandler::init(IOBufferReader * buffer_in, HttpTunnelProducer * p)
 {
+  if (p->do_chunking)
+    init_by_action(buffer_in, ACTION_DOCHUNK);
+  else if (p->do_dechunking)
+    init_by_action(buffer_in, ACTION_DECHUNK);
+  else
+    init_by_action(buffer_in, ACTION_PASSTHRU);
+  return;
+}
+
+void
+ChunkedHandler::init_by_action(IOBufferReader *buffer_in, Action action)
+{
   running_sum = 0;
   num_digits = 0;
   cur_chunk_size = 0;
   bytes_left = 0;
   truncation = false;
+  this->action = action;
 
-  if (p->do_chunking) {
+  switch (action) {
+  case ACTION_DOCHUNK:
     dechunked_reader = buffer_in->mbuf->clone_reader(buffer_in);
     dechunked_reader->mbuf->water_mark = min_block_transfer_bytes;
     chunked_buffer = new_MIOBuffer(CHUNK_IOBUFFER_SIZE_INDEX);
     chunked_size = 0;
-  } else {
-    ink_assert(p->do_dechunking || p->do_chunked_passthru);
+    break;
+  case ACTION_DECHUNK:
+    chunked_reader = buffer_in->mbuf->clone_reader(buffer_in);
+    dechunked_buffer = new_MIOBuffer(BUFFER_SIZE_INDEX_256);
+    dechunked_size = 0;
+    break;
+  case ACTION_PASSTHRU:
     chunked_reader = buffer_in->mbuf->clone_reader(buffer_in);
+    break;
+  default:
+    ink_release_assert(!"Unknown action");
+  }
 
-    if (p->do_dechunking) {
-      // This is the min_block_transfer_bytes value.
-      dechunked_buffer = new_MIOBuffer(BUFFER_SIZE_INDEX_256);
-      dechunked_size = 0;
-    }
+  return;
+}
+
+void
+ChunkedHandler::clear()
+{
+  switch (action) {
+  case ACTION_DOCHUNK:
+    free_MIOBuffer(chunked_buffer);
+    break;
+  case ACTION_DECHUNK:
+    free_MIOBuffer(dechunked_buffer);
+    break;
+  case ACTION_PASSTHRU:
+  default:
+    break;
   }
+
+  return;
 }
 
 void

http://git-wip-us.apache.org/repos/asf/trafficserver/blob/8a0bee4e/proxy/http/HttpTunnel.h
----------------------------------------------------------------------
diff --git a/proxy/http/HttpTunnel.h b/proxy/http/HttpTunnel.h
index e4bdd71..81fac50 100644
--- a/proxy/http/HttpTunnel.h
+++ b/proxy/http/HttpTunnel.h
@@ -86,8 +86,7 @@ enum TunnelChunkingAction_t
 
 struct ChunkedHandler
 {
-  enum ChunkedState
-  {
+  enum ChunkedState {
     CHUNK_READ_CHUNK = 0,
     CHUNK_READ_SIZE_START,
     CHUNK_READ_SIZE,
@@ -104,6 +103,14 @@ struct ChunkedHandler
 
   static int const DEFAULT_MAX_CHUNK_SIZE = 4096;
 
+  enum Action {
+    ACTION_DOCHUNK = 0,
+    ACTION_DECHUNK,
+    ACTION_PASSTHRU,
+  };
+
+  Action action;
+
   IOBufferReader *chunked_reader;
   MIOBuffer *dechunked_buffer;
   int64_t dechunked_size;
@@ -137,7 +144,9 @@ struct ChunkedHandler
   //@}
   ChunkedHandler();
 
-  void init(IOBufferReader * buffer_in, HttpTunnelProducer * p);
+  void init(IOBufferReader *buffer_in, HttpTunnelProducer *p);
+  void init_by_action(IOBufferReader *buffer_in, Action action);
+  void clear();
 
   /// Set the max chunk @a size.
   /// If @a size is zero it is set to @c DEFAULT_MAX_CHUNK_SIZE.