message: tune buffer size, add non blocking logic in string mutex aquiring
diff: @@ -24,7 +24,7 @@ impl<'a> FromPyObject<'a> for StringBlock<'a> {
  Ok(StringBlock {
  data,
  mutex: Arc::new(Mutex::new(())), // allocate the lock here since only StringBlock needs to aquire the GIL for now
  - buf_size_mb: 16, // in MB
  + buf_size_mb: 4, // in MB
  })
  }
  }
  @@ -76,7 +76,7 @@ impl<'a> PandasColumnObject for StringColumn<'a> {
  }
  #[throws(ConnectorAgentError)]
  fn finalize(&mut self) {
  - self.flush()?;
  + self.flush(true)?;
  }
  }
  @@ -244,19 +244,21 @@ impl<'a> StringColumn<'a> {
  }
  #[throws(ConnectorAgentError)]
  - pub fn flush(&mut self) {
  + pub fn flush(&mut self, force_flush: bool) {
  let nstrings = self.string_lengths.len();
  + if nstrings == 0 {
  + return;
  + }
  - if nstrings > 0 {
  let py = unsafe { Python::assume_gil_acquired() };
  - let mut string_infos = Vec::with_capacity(self.string_lengths.len());
  -
  - {
  + let string_infos = match force_flush {
  + true => {
  // allocation in python is not thread safe
  let _guard = self
  .mutex
  .lock()
  .map_err(|e| anyhow!("mutex poisoned {}", e))?;
  + let mut string_infos = Vec::with_capacity(self.string_lengths.len());
  let mut start = 0;
  for (i, &len) in self.string_lengths.iter().enumerate() {
  let end = start + len;
  @@ -273,8 +275,36 @@ impl<'a> StringColumn<'a> {
  }
  start = end;
  }
  + string_infos
  }
  + false => match self.mutex.try_lock() {
  + Ok(_guard) => {
  + let mut string_infos = Vec::with_capacity(self.string_lengths.len());
  + let mut start = 0;
  + for (i, &len) in self.string_lengths.iter().enumerate() {
  + let end = start + len;
  + unsafe {
  + string_infos.push(StringInfo::detect(&self.string_buf[start..end]));
  + }
  + if len != 0 {
  + unsafe {
  + *self.data.get_unchecked_mut(self.next_write + i) = string_infos
  + .last()
  + .ok_or_else(|| anyhow!("empty string_info vector"))?
  + .pystring(py)
  + };
  + }
  + start = end;
  + }
  + string_infos
  + }
  + Err(_e) => {
  + vec![]
  + }
  + },
  + };
  + if string_infos.len() > 0 {
  let mut start = 0;
  for (i, (len, info)) in self
  .string_lengths
  @@ -299,7 +329,9 @@ impl<'a> StringColumn<'a> {
  #[throws(ConnectorAgentError)]
  pub fn try_flush(&mut self) {
  if self.string_buf.len() >= self.buf_size {
  - self.flush()?;
  + self.flush(true)?;
  + } else if self.string_buf.len() >= self.buf_size / 2 {
  + self.flush(false)?;
  }
  }
  }
  